diff --git a/.github/workflows/build_and_test.yaml b/.github/workflows/build_and_test.yaml index 3094cf72c..e73469b5c 100644 --- a/.github/workflows/build_and_test.yaml +++ b/.github/workflows/build_and_test.yaml @@ -71,6 +71,10 @@ jobs: if: ${{ !matrix.skip_rust }} shell: bash run: ci/scripts/setup_rust.sh + - name: Install HTTP and TLS development dependencies + run: | + sudo apt-get update + sudo apt-get install -y libcurl4-openssl-dev libssl-dev - name: Build Paimon shell: bash env: diff --git a/.github/workflows/gcc8_test.yaml b/.github/workflows/gcc8_test.yaml index c4316b032..2a30e2ef4 100644 --- a/.github/workflows/gcc8_test.yaml +++ b/.github/workflows/gcc8_test.yaml @@ -42,7 +42,7 @@ jobs: - name: Install dependencies run: | apt-get update - DEBIAN_FRONTEND=noninteractive apt-get install -y gcc-8 g++-8 ninja-build git git-lfs tar curl tzdata zip unzip pkg-config build-essential python3-dev gdb sudo + DEBIAN_FRONTEND=noninteractive apt-get install -y gcc-8 g++-8 ninja-build git git-lfs tar curl libcurl4-openssl-dev libssl-dev tzdata zip unzip pkg-config build-essential python3-dev gdb sudo curl -L -O https://github.com/Kitware/CMake/releases/download/v3.28.3/cmake-3.28.3-linux-x86_64.tar.gz tar -zxvf cmake-3.28.3-linux-x86_64.tar.gz -C /usr/local --strip-components=1 rm cmake-3.28.3-linux-x86_64.tar.gz diff --git a/CMakeLists.txt b/CMakeLists.txt index da9bbd652..c1b6c771f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -57,6 +57,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_LANCE "Whether to enable lance file format" OFF) option(PAIMON_ENABLE_JINDO "Whether to enable jindo 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) option(PAIMON_ENABLE_LUMINA "Whether to enable lumina vector index" OFF) @@ -72,6 +73,9 @@ endif() if(PAIMON_ENABLE_JINDO) add_definitions(-DPAIMON_ENABLE_JINDO) endif() +if(PAIMON_ENABLE_S3) + add_definitions(-DPAIMON_ENABLE_S3) +endif() if(PAIMON_ENABLE_NETWORK_TESTS) if(NOT PAIMON_BUILD_TESTS) message(FATAL_ERROR "PAIMON_ENABLE_NETWORK_TESTS requires PAIMON_BUILD_TESTS=ON") @@ -455,6 +459,13 @@ if(PAIMON_BUILD_TESTS) paimon_jindo_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) + paimon_link_libraries_no_as_needed(TEST_PLUGIN_LINK_LIBS + paimon_s3_file_system_shared) + list(APPEND TEST_STATIC_LINK_LIBS ${TEST_PLUGIN_LINK_LIBS}) + endif() if(PAIMON_ENABLE_LUMINA) paimon_link_libraries_whole_archive(PAIMON_LUMINA_INDEX_STATIC_LINK_LIBS paimon_lumina_index_static) @@ -505,6 +516,11 @@ config_summary_message() add_subdirectory(src/paimon) add_subdirectory(src/paimon/fs/local) add_subdirectory(src/paimon/fs/jindo) +if(PAIMON_ENABLE_S3) + find_package(CURL REQUIRED) +endif() +add_subdirectory(src/paimon/fs/object_store) +add_subdirectory(src/paimon/fs/s3) add_subdirectory(src/paimon/format/blob) add_subdirectory(src/paimon/format/orc) add_subdirectory(src/paimon/format/parquet) diff --git a/ci/scripts/build_paimon.sh b/ci/scripts/build_paimon.sh index ba022423c..8125ee973 100755 --- a/ci/scripts/build_paimon.sh +++ b/ci/scripts/build_paimon.sh @@ -133,6 +133,7 @@ CMAKE_ARGS=( "-DPAIMON_BUILD_TESTS=ON" "-DPAIMON_ENABLE_LANCE=${ENABLE_LANCE}" "-DPAIMON_ENABLE_JINDO=ON" + "-DPAIMON_ENABLE_S3=ON" "-DPAIMON_ENABLE_LUMINA=${ENABLE_LUMINA}" "-DPAIMON_ENABLE_LUCENE=ON" "-DPAIMON_ENABLE_TANTIVY=${ENABLE_TANTIVY}" diff --git a/cmake_modules/BuildAwsAuth.cmake b/cmake_modules/BuildAwsAuth.cmake new file mode 100644 index 000000000..0b022ff18 --- /dev/null +++ b/cmake_modules/BuildAwsAuth.cmake @@ -0,0 +1,187 @@ +# Copyright 2026-present Alibaba Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +include(ExternalProject) +include(GNUInstallDirs) + +function(paimon_add_s2n_project) + if(DEFINED ENV{PAIMON_AWS_S2N_URL}) + set(S2N_URL "$ENV{PAIMON_AWS_S2N_URL}") + elseif(EXISTS "${THIRDPARTY_DIR}/${PAIMON_AWS_S2N_PKG_NAME}") + set(S2N_URL "${THIRDPARTY_DIR}/${PAIMON_AWS_S2N_PKG_NAME}") + else() + set(S2N_URL + "${THIRDPARTY_MIRROR_URL}https://codeload.github.com/awslabs/s2n/zip/${PAIMON_AWS_S2N_BUILD_REVISION}" + ) + endif() + externalproject_add(s2n_ep + URL ${S2N_URL} + URL_HASH SHA256=${PAIMON_AWS_S2N_BUILD_SHA256_CHECKSUM} + DOWNLOAD_EXTRACT_TIMESTAMP TRUE + CMAKE_ARGS ${EP_COMMON_TOOLCHAIN} + -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE} + -DCMAKE_INSTALL_PREFIX=${AWS_AUTH_PREFIX} + -DCMAKE_INSTALL_LIBDIR=${AWS_AUTH_INSTALL_LIBDIR} + -DCMAKE_PREFIX_PATH=${AWS_AUTH_PREFIX} + -Dcrypto_INCLUDE_DIR=${OPENSSL_INCLUDE_DIR} + -Dcrypto_LIBRARY=${OPENSSL_CRYPTO_LIBRARY} + -DBUILD_SHARED_LIBS=OFF + -DCMAKE_POSITION_INDEPENDENT_CODE=ON + -DBUILD_TESTING=OFF + -DS2N_INTERN_LIBCRYPTO=OFF + BUILD_BYPRODUCTS ${AWS_AUTH_LIB_DIR}/libs2n.a + ${THIRDPARTY_LOG_OPTIONS}) +endfunction() + +function(paimon_add_aws_c_project + NAME + VERSION + CHECKSUM + DEPENDS) + string(REPLACE "aws-c-" "AWS_C_" URL_NAME "${NAME}") + string(TOUPPER "${URL_NAME}" URL_NAME) + set(URL_VAR "PAIMON_${URL_NAME}_URL") + set(PKG_NAME_VAR "PAIMON_${URL_NAME}_PKG_NAME") + if(DEFINED ENV{${URL_VAR}}) + set(URL "$ENV{${URL_VAR}}") + elseif(EXISTS "${THIRDPARTY_DIR}/${${PKG_NAME_VAR}}") + set(URL "${THIRDPARTY_DIR}/${${PKG_NAME_VAR}}") + else() + set(URL + "${THIRDPARTY_MIRROR_URL}https://github.com/awslabs/${NAME}/archive/refs/tags/${VERSION}.tar.gz" + ) + endif() + set(AWS_PLATFORM_CMAKE_ARGS) + list(APPEND + AWS_PLATFORM_CMAKE_ARGS + -DOPENSSL_INCLUDE_DIR=${OPENSSL_INCLUDE_DIR} + -DOPENSSL_SSL_LIBRARY=${OPENSSL_SSL_LIBRARY} + -DOPENSSL_CRYPTO_LIBRARY=${OPENSSL_CRYPTO_LIBRARY} + -Dcrypto_INCLUDE_DIR=${OPENSSL_INCLUDE_DIR} + -Dcrypto_LIBRARY=${OPENSSL_CRYPTO_LIBRARY}) + if(NAME STREQUAL "aws-c-cal") + list(APPEND AWS_PLATFORM_CMAKE_ARGS -DUSE_OPENSSL=ON) + endif() + if(APPLE) + list(APPEND AWS_PLATFORM_CMAKE_ARGS -DAWS_USE_SECITEM=ON) + endif() + # aws-c-io's installed package unconditionally looks for s2n on Unix even + # when it was built with Apple's native TLS backend. + if(APPLE AND (NAME STREQUAL "aws-c-http" OR NAME STREQUAL "aws-c-auth")) + list(APPEND AWS_PLATFORM_CMAKE_ARGS -DBYO_CRYPTO=ON) + endif() + externalproject_add(${NAME}_ep + URL ${URL} + URL_HASH SHA256=${CHECKSUM} + DOWNLOAD_EXTRACT_TIMESTAMP TRUE + CMAKE_ARGS ${EP_COMMON_TOOLCHAIN} + -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE} + -DCMAKE_INSTALL_PREFIX=${AWS_AUTH_PREFIX} + -DCMAKE_INSTALL_LIBDIR=${AWS_AUTH_INSTALL_LIBDIR} + -DCMAKE_PREFIX_PATH=${AWS_AUTH_PREFIX} + -DBUILD_SHARED_LIBS=OFF + -DCMAKE_POSITION_INDEPENDENT_CODE=ON + -DBUILD_TESTING=OFF + -DENABLE_TESTING=OFF + ${AWS_PLATFORM_CMAKE_ARGS} + DEPENDS ${DEPENDS} + BUILD_BYPRODUCTS ${AWS_AUTH_LIB_DIR}/lib${NAME}.a + ${THIRDPARTY_LOG_OPTIONS}) +endfunction() + +function(build_aws_auth) + set(AWS_AUTH_PREFIX + "${CMAKE_CURRENT_BINARY_DIR}/aws-auth_ep-install" + PARENT_SCOPE) + set(AWS_AUTH_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/aws-auth_ep-install") + set(AWS_AUTH_INSTALL_LIBDIR "${CMAKE_INSTALL_LIBDIR}") + set(AWS_AUTH_LIB_DIR "${AWS_AUTH_PREFIX}/${AWS_AUTH_INSTALL_LIBDIR}") + file(MAKE_DIRECTORY "${AWS_AUTH_PREFIX}/include") + file(MAKE_DIRECTORY "${AWS_AUTH_LIB_DIR}") + + find_package(OpenSSL REQUIRED) + + set(AWS_IO_DEPENDS "aws-c-common_ep;aws-c-cal_ep") + if(CMAKE_SYSTEM_NAME STREQUAL "Linux") + paimon_add_s2n_project() + list(APPEND AWS_IO_DEPENDS s2n_ep) + endif() + + paimon_add_aws_c_project(aws-c-common ${PAIMON_AWS_C_COMMON_BUILD_VERSION} + ${PAIMON_AWS_C_COMMON_BUILD_SHA256_CHECKSUM} "") + paimon_add_aws_c_project(aws-c-sdkutils ${PAIMON_AWS_C_SDKUTILS_BUILD_VERSION} + ${PAIMON_AWS_C_SDKUTILS_BUILD_SHA256_CHECKSUM} + aws-c-common_ep) + paimon_add_aws_c_project(aws-c-cal ${PAIMON_AWS_C_CAL_BUILD_VERSION} + ${PAIMON_AWS_C_CAL_BUILD_SHA256_CHECKSUM} aws-c-common_ep) + paimon_add_aws_c_project(aws-c-compression ${PAIMON_AWS_C_COMPRESSION_BUILD_VERSION} + ${PAIMON_AWS_C_COMPRESSION_BUILD_SHA256_CHECKSUM} + aws-c-common_ep) + paimon_add_aws_c_project(aws-c-io ${PAIMON_AWS_C_IO_BUILD_VERSION} + ${PAIMON_AWS_C_IO_BUILD_SHA256_CHECKSUM} "${AWS_IO_DEPENDS}") + paimon_add_aws_c_project(aws-c-http ${PAIMON_AWS_C_HTTP_BUILD_VERSION} + ${PAIMON_AWS_C_HTTP_BUILD_SHA256_CHECKSUM} + "aws-c-common_ep;aws-c-io_ep;aws-c-compression_ep") + paimon_add_aws_c_project(aws-c-auth + ${PAIMON_AWS_C_AUTH_BUILD_VERSION} + ${PAIMON_AWS_C_AUTH_BUILD_SHA256_CHECKSUM} + "aws-c-common_ep;aws-c-sdkutils_ep;aws-c-cal_ep;aws-c-io_ep;aws-c-http_ep" + ) + + find_package(Threads REQUIRED) + set(AWS_AUTH_PLATFORM_LIBS Threads::Threads ${CMAKE_DL_LIBS}) + if(APPLE) + list(APPEND + AWS_AUTH_PLATFORM_LIBS + "-framework Security" + "-framework CoreFoundation" + "-framework Network") + endif() + set(AWS_AUTH_TLS_LIBS) + if(CMAKE_SYSTEM_NAME STREQUAL "Linux") + list(APPEND AWS_AUTH_TLS_LIBS "${AWS_AUTH_LIB_DIR}/libs2n.a") + endif() + set(AWS_AUTH_RUNTIME_LIBS + "${AWS_AUTH_LIB_DIR}/libaws-c-http.a" + "${AWS_AUTH_LIB_DIR}/libaws-c-io.a" + "${AWS_AUTH_LIB_DIR}/libaws-c-compression.a" + "${AWS_AUTH_LIB_DIR}/libaws-c-cal.a" + "${AWS_AUTH_LIB_DIR}/libaws-c-sdkutils.a" + "${AWS_AUTH_LIB_DIR}/libaws-c-common.a" + ${AWS_AUTH_TLS_LIBS}) + if(NOT APPLE) + list(PREPEND AWS_AUTH_RUNTIME_LIBS "-Wl,--start-group") + list(APPEND AWS_AUTH_RUNTIME_LIBS "-Wl,--end-group") + endif() + # Keep OpenSSL after the static AWS archives. In embedded builds CMake can + # resolve these targets to static archives, where link order matters. + list(APPEND AWS_AUTH_RUNTIME_LIBS "${OPENSSL_SSL_LIBRARY}" + "${OPENSSL_CRYPTO_LIBRARY}") + add_library(aws_auth_minimal STATIC IMPORTED GLOBAL) + set_target_properties(aws_auth_minimal + PROPERTIES IMPORTED_LOCATION + "${AWS_AUTH_LIB_DIR}/libaws-c-auth.a" + INTERFACE_INCLUDE_DIRECTORIES + "${AWS_AUTH_PREFIX}/include" + INTERFACE_LINK_LIBRARIES + "${AWS_AUTH_RUNTIME_LIBS};${AWS_AUTH_PLATFORM_LIBS}") + add_dependencies(aws_auth_minimal aws-c-auth_ep) + + set(AWS_AUTH_INCLUDE_DIR + "${AWS_AUTH_PREFIX}/include" + PARENT_SCOPE) + set(AWS_AUTH_LIB_DIR + "${AWS_AUTH_LIB_DIR}" + PARENT_SCOPE) +endfunction() diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake index 94bc1de57..ad489ba3c 100644 --- a/cmake_modules/ThirdpartyToolchain.cmake +++ b/cmake_modules/ThirdpartyToolchain.cmake @@ -49,6 +49,7 @@ foreach(_VERSION_ENTRY ${TOOLCHAIN_VERSIONS_TXT}) if(NOT ((_VERSION_ENTRY MATCHES "^[^#][A-Za-z0-9-_]+_VERSION=") OR (_VERSION_ENTRY MATCHES "^[^#][A-Za-z0-9-_]+_CHECKSUM=") + OR (_VERSION_ENTRY MATCHES "^[^#][A-Za-z0-9-_]+_REVISION=") OR (_VERSION_ENTRY MATCHES "^[^#][A-Za-z0-9-_]+_PKG_NAME="))) continue() endif() @@ -1877,6 +1878,10 @@ if(PAIMON_ENABLE_JINDO) build_jindosdk_c() build_jindosdk_nextarch() endif() +if(PAIMON_ENABLE_S3) + include(BuildAwsAuth) + build_aws_auth() +endif() if(PAIMON_ENABLE_LUMINA) build_lumina() endif() diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff index ae7752097..c50b00c1c 100644 --- a/cmake_modules/arrow.diff +++ b/cmake_modules/arrow.diff @@ -415,6 +415,14 @@ diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake b/cpp/cmake_modules/Thi # and crosscompiling emulator (for try_run() ) if(CMAKE_CROSSCOMPILING_EMULATOR) string(REPLACE ";" ${EP_LIST_SEPARATOR} EP_CMAKE_CROSSCOMPILING_EMULATOR +@@ -1720,6 +1725,7 @@ macro(build_thrift) + -DWITH_JAVASCRIPT=OFF + -DWITH_LIBEVENT=OFF + -DWITH_NODEJS=OFF ++ -DWITH_OPENSSL=OFF + -DWITH_PYTHON=OFF + -DWITH_QT5=OFF + -DWITH_ZLIB=OFF) diff --git a/cpp/cmake_modules/BuildUtils.cmake b/cpp/cmake_modules/BuildUtils.cmake --- a/cpp/cmake_modules/BuildUtils.cmake +++ b/cpp/cmake_modules/BuildUtils.cmake diff --git a/src/paimon/fs/object_store/CMakeLists.txt b/src/paimon/fs/object_store/CMakeLists.txt new file mode 100644 index 000000000..8c9c85e92 --- /dev/null +++ b/src/paimon/fs/object_store/CMakeLists.txt @@ -0,0 +1,41 @@ +# Copyright 2026-present Alibaba Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +if(PAIMON_ENABLE_S3) + add_paimon_lib(paimon_object_store + SOURCES + http_client.cpp + object_store_file_system.cpp + DEPENDENCIES + paimon_shared + CURL::libcurl + STATIC_LINK_LIBS + CURL::libcurl + fmt + SHARED_LINK_LIBS + paimon_shared + SHARED_LINK_FLAGS + ${PAIMON_VERSION_SCRIPT_FLAGS}) + + if(PAIMON_BUILD_TESTS) + add_paimon_test(object_store_file_system_test + SOURCES + object_store_file_system_test.cpp + STATIC_LINK_LIBS + paimon_shared + test_utils_static + paimon_object_store_static + ${GTEST_LINK_TOOLCHAIN}) + endif() +endif() diff --git a/src/paimon/fs/object_store/http_client.cpp b/src/paimon/fs/object_store/http_client.cpp new file mode 100644 index 000000000..35689cdc4 --- /dev/null +++ b/src/paimon/fs/object_store/http_client.cpp @@ -0,0 +1,200 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/fs/object_store/http_client.h" + +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "fmt/format.h" +#include "paimon/common/utils/string_utils.h" + +namespace paimon::object_store { + +namespace { + +constexpr int kMaxAttempts = 3; + +class CurlGlobalGuard { + public: + CurlGlobalGuard() { + curl_global_init(CURL_GLOBAL_DEFAULT); + } + ~CurlGlobalGuard() { + curl_global_cleanup(); + } +}; + +std::shared_ptr GetCurlGlobalGuard() { + static auto guard = std::make_shared(); + return guard; +} + +std::string Trim(std::string value) { + auto not_space = [](unsigned char c) { return !std::isspace(c); }; + value.erase(value.begin(), std::find_if(value.begin(), value.end(), not_space)); + value.erase(std::find_if(value.rbegin(), value.rend(), not_space).base(), value.end()); + return value; +} + +struct TransferContext { + CURL* handle; + const HttpBodyConsumer* consumer; + HttpResponse response; + Status status = Status::OK(); +}; + +size_t WriteCallback(char* data, size_t size, size_t count, void* user_data) { + auto* context = static_cast(user_data); + size_t bytes = size * count; + if (!context->status.ok()) { + return 0; + } + long status_code = 0; // NOLINT(runtime/int, google-runtime-int): required by curl. + curl_easy_getinfo(context->handle, CURLINFO_RESPONSE_CODE, &status_code); + context->response.status_code = static_cast(status_code); + if (status_code >= 300) { + context->response.body_size += static_cast(bytes); + return bytes; + } + context->status = (*context->consumer)(data, static_cast(bytes)); + if (!context->status.ok()) { + return 0; + } + context->response.body_size += static_cast(bytes); + return bytes; +} + +size_t HeaderCallback(char* data, size_t size, size_t count, void* user_data) { + auto* context = static_cast(user_data); + size_t bytes = size * count; + std::string line(data, bytes); + size_t colon = line.find(':'); + if (colon != std::string::npos) { + std::string name = Trim(line.substr(0, colon)); + name = StringUtils::ToLowerCase(name); + context->response.headers[name] = Trim(line.substr(colon + 1)); + } + return bytes; +} + +bool IsRetryable(CURLcode code, int64_t status_code) { + if (code == CURLE_COULDNT_RESOLVE_HOST || code == CURLE_COULDNT_CONNECT || + code == CURLE_OPERATION_TIMEDOUT || code == CURLE_SEND_ERROR || code == CURLE_RECV_ERROR || + code == CURLE_PARTIAL_FILE) { + return true; + } + return status_code == 429 || status_code >= 500; +} + +} // namespace + +class CurlHttpClient::Impl { + public: + Impl() : guard_(GetCurlGlobalGuard()) {} + + ~Impl() { + for (CURL* handle : handles_) { + curl_easy_cleanup(handle); + } + } + + CURL* Acquire() const { + std::scoped_lock lock(mutex_); + if (handles_.empty()) { + return curl_easy_init(); + } + CURL* handle = handles_.back(); + handles_.pop_back(); + return handle; + } + + void Release(CURL* handle) const { + curl_easy_reset(handle); + std::scoped_lock lock(mutex_); + handles_.push_back(handle); + } + + private: + std::shared_ptr guard_; + mutable std::mutex mutex_; + mutable std::vector handles_; +}; + +CurlHttpClient::CurlHttpClient() : impl_(std::make_unique()) {} +CurlHttpClient::~CurlHttpClient() = default; + +Result CurlHttpClient::Execute(const HttpRequest& request, + const HttpBodyConsumer& consumer) const { + for (int attempt = 0; attempt < kMaxAttempts; ++attempt) { + CURL* handle = impl_->Acquire(); + if (handle == nullptr) { + return Status::IOError("failed to create curl easy handle"); + } + TransferContext context{handle, &consumer, {}, Status::OK()}; + curl_slist* headers = nullptr; + for (const auto& [name, value] : request.headers) { + headers = curl_slist_append(headers, (name + ": " + value).c_str()); + } + curl_easy_setopt(handle, CURLOPT_URL, request.url.c_str()); + curl_easy_setopt(handle, CURLOPT_HTTPHEADER, headers); + curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L); + curl_easy_setopt(handle, CURLOPT_CONNECTTIMEOUT_MS, 30000L); + curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, WriteCallback); + curl_easy_setopt(handle, CURLOPT_WRITEDATA, &context); + curl_easy_setopt(handle, CURLOPT_HEADERFUNCTION, HeaderCallback); + curl_easy_setopt(handle, CURLOPT_HEADERDATA, &context); + curl_easy_setopt(handle, CURLOPT_FOLLOWLOCATION, 0L); + if (request.method == HttpMethod::HEAD) { + curl_easy_setopt(handle, CURLOPT_NOBODY, 1L); + } + CURLcode code = curl_easy_perform(handle); + long response_code = 0; // NOLINT(runtime/int, google-runtime-int): required by curl. + curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code); + context.response.status_code = static_cast(response_code); + curl_slist_free_all(headers); + impl_->Release(handle); + + if (!context.status.ok()) { + return context.status; + } + if (code == CURLE_OK && !IsRetryable(code, response_code)) { + return context.response; + } + if (!IsRetryable(code, response_code) || + (context.response.body_size > 0 && response_code < 300) || + attempt + 1 == kMaxAttempts) { + if (code == CURLE_OK) { + return Status::IOError(fmt::format("HTTP request to {} returned status {}", + request.url, response_code)); + } + return Status::IOError(fmt::format("HTTP request to {} failed: {} (status {})", + request.url, curl_easy_strerror(code), + response_code)); + } + std::this_thread::sleep_for(std::chrono::milliseconds(100 * (1 << attempt))); + } + return Status::IOError("HTTP request failed"); +} + +} // namespace paimon::object_store diff --git a/src/paimon/fs/object_store/http_client.h b/src/paimon/fs/object_store/http_client.h new file mode 100644 index 000000000..ee5a65cc6 --- /dev/null +++ b/src/paimon/fs/object_store/http_client.h @@ -0,0 +1,66 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include + +#include "paimon/result.h" +#include "paimon/visibility.h" + +namespace paimon::object_store { + +enum class HttpMethod { HEAD, GET }; +using HttpHeaders = std::map; +using HttpBodyConsumer = std::function; + +struct HttpRequest { + HttpMethod method = HttpMethod::GET; + std::string url; + HttpHeaders headers; +}; + +struct HttpResponse { + int status_code = 0; + HttpHeaders headers; + int64_t body_size = 0; +}; + +class PAIMON_EXPORT HttpClient { + public: + virtual ~HttpClient() = default; + virtual Result Execute(const HttpRequest& request, + const HttpBodyConsumer& consumer) const = 0; +}; + +class PAIMON_EXPORT CurlHttpClient : public HttpClient { + public: + CurlHttpClient(); + ~CurlHttpClient() override; + + Result Execute(const HttpRequest& request, + const HttpBodyConsumer& consumer) const override; + + private: + class Impl; + std::unique_ptr impl_; +}; + +} // namespace paimon::object_store diff --git a/src/paimon/fs/object_store/object_store_file_system.cpp b/src/paimon/fs/object_store/object_store_file_system.cpp new file mode 100644 index 000000000..db50bad5f --- /dev/null +++ b/src/paimon/fs/object_store/object_store_file_system.cpp @@ -0,0 +1,532 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/fs/object_store/object_store_file_system.h" + +#include +#include +#include +#include +#include + +#include "fmt/format.h" +#include "paimon/common/utils/math.h" +#include "paimon/common/utils/path_util.h" + +namespace paimon::object_store { +namespace { + +constexpr int64_t kMaxReadAheadMemory = 64LL * 1024LL * 1024LL; +constexpr int64_t kInitialReadAheadSize = 64LL * 1024LL; +constexpr int64_t kMaxReadAheadSize = 8LL * 1024LL * 1024LL; + +std::string NormalizeDirectoryPrefix(const std::string& key) { + if (key.empty() || key.back() == '/') { + return key; + } + return key + "/"; +} + +class ObjectStoreBasicFileStatus : public BasicFileStatus { + public: + ObjectStoreBasicFileStatus(std::string path, bool is_dir) + : path_(std::move(path)), is_dir_(is_dir) {} + bool IsDir() const override { + return is_dir_; + } + std::string GetPath() const override { + return path_; + } + + private: + std::string path_; + bool is_dir_; +}; + +class ObjectStoreFileStatus : public FileStatus { + public: + ObjectStoreFileStatus(std::string path, int64_t size, int64_t modification_time, bool is_dir) + : path_(std::move(path)), + size_(size), + modification_time_(modification_time), + is_dir_(is_dir) {} + int64_t GetLen() const override { + return size_; + } + bool IsDir() const override { + return is_dir_; + } + std::string GetPath() const override { + return path_; + } + int64_t GetModificationTime() const override { + return modification_time_; + } + + private: + std::string path_; + int64_t size_; + int64_t modification_time_; + bool is_dir_; +}; + +class ObjectStoreInputStream : public InputStream { + public: + ObjectStoreInputStream(std::shared_ptr client, + std::shared_ptr limiter, ObjectStorePath path, + std::string uri, int64_t length) + : client_(std::move(client)), + limiter_(std::move(limiter)), + path_(std::move(path)), + uri_(std::move(uri)), + length_(length) {} + + ~ObjectStoreInputStream() override { + CloseInternal(); + } + + Status Seek(int64_t offset, SeekOrigin origin) override { + std::scoped_lock lock(mutex_); + if (closed_) { + return Status::IOError(fmt::format("{} is closed", uri_)); + } + int64_t base; + if (origin == FS_SEEK_SET) { + base = 0; + } else if (origin == FS_SEEK_CUR) { + base = position_; + } else if (origin == FS_SEEK_END) { + base = length_; + } else { + return Status::Invalid("unsupported seek origin"); + } + if ((offset > 0 && base > std::numeric_limits::max() - offset) || + (offset < 0 && base < std::numeric_limits::min() - offset)) { + return Status::Invalid("object store input position overflows"); + } + int64_t position = base + offset; + PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(position, "object store input position")); + position_ = position; + return Status::OK(); + } + + Result GetPos() const override { + std::scoped_lock lock(mutex_); + if (closed_) { + return Status::IOError(fmt::format("{} is closed", uri_)); + } + return position_; + } + + Result Read(char* buffer, int64_t size) override { + std::scoped_lock read_lock(read_mutex_); + int64_t offset; + { + std::scoped_lock lock(mutex_); + if (closed_) { + return Status::IOError(fmt::format("{} is closed", uri_)); + } + offset = position_; + } + PAIMON_ASSIGN_OR_RAISE(int64_t bytes_read, ReadWithReadAhead(buffer, size, offset)); + { + std::scoped_lock lock(mutex_); + position_ += bytes_read; + } + return bytes_read; + } + + Result Read(char* buffer, int64_t size, int64_t offset) override { + std::scoped_lock read_lock(read_mutex_); + return ReadWithReadAhead(buffer, size, offset); + } + + void ReadAsync(char* buffer, int64_t size, int64_t offset, + std::function&& callback) override { + Status status = ValidateRead(buffer, size, offset); + if (!status.ok()) { + callback(std::move(status)); + return; + } + if (offset > length_ || size > length_ - offset) { + callback(Status::Invalid( + fmt::format("object store async read size {} at offset {} exceeds length {}", size, + offset, length_))); + return; + } + if (size == 0) { + callback(Status::OK()); + return; + } + client_->GetObjectRangeAsync(path_, offset, size, buffer, + [callback = std::move(callback)](Status status) mutable { + if (!status.ok()) { + callback(std::move(status)); + return; + } + callback(Status::OK()); + }); + } + + Status Close() override { + CloseInternal(); + return Status::OK(); + } + Result GetUri() const override { + return uri_; + } + Result Length() const override { + return length_; + } + + private: + Status ValidateRead(char* buffer, int64_t size, int64_t offset) const { + PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(size, "read length")); + PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(offset, "read offset")); + { + std::scoped_lock lock(mutex_); + if (closed_) { + return Status::IOError(fmt::format("{} is closed", uri_)); + } + } + if (size > 0 && buffer == nullptr) { + return Status::Invalid("read buffer is null"); + } + return Status::OK(); + } + + void EvictReadAhead() { + int64_t reserved = 0; + { + std::scoped_lock lock(mutex_); + std::vector().swap(read_ahead_buffer_); + reserved = reserved_read_ahead_size_; + reserved_read_ahead_size_ = 0; + } + limiter_->Release(reserved); + } + + Result ReadWithReadAhead(char* buffer, int64_t size, int64_t offset) { + PAIMON_RETURN_NOT_OK(ValidateRead(buffer, size, offset)); + if (size == 0 || offset >= length_) { + return 0; + } + int64_t read_size = std::min(size, length_ - offset); + bool continues_previous = false; + { + std::scoped_lock lock(mutex_); + int64_t relative = offset - read_ahead_offset_; + if (relative >= 0 && relative <= static_cast(read_ahead_buffer_.size()) && + read_size <= static_cast(read_ahead_buffer_.size()) - relative) { + std::memcpy(buffer, read_ahead_buffer_.data() + relative, + static_cast(read_size)); + return read_size; + } + continues_previous = + !read_ahead_buffer_.empty() && + offset == read_ahead_offset_ + static_cast(read_ahead_buffer_.size()); + } + EvictReadAhead(); + if (read_size >= kMaxReadAheadSize) { + next_read_ahead_size_ = kInitialReadAheadSize; + return client_->GetObjectRange(path_, offset, read_size, buffer); + } + next_read_ahead_size_ = continues_previous + ? std::min(kMaxReadAheadSize, next_read_ahead_size_ * 2) + : kInitialReadAheadSize; + int64_t fetch_size = std::min(std::max(read_size, next_read_ahead_size_), length_ - offset); + fetch_size = limiter_->ReserveUpTo(read_size, fetch_size); + if (fetch_size == 0) { + return client_->GetObjectRange(path_, offset, read_size, buffer); + } + std::vector fetched(static_cast(fetch_size)); + Result result = client_->GetObjectRange(path_, offset, fetch_size, fetched.data()); + if (!result.ok()) { + limiter_->Release(fetch_size); + return result.status(); + } + int64_t bytes_read = std::move(result).value(); + if (bytes_read != fetch_size) { + limiter_->Release(fetch_size); + return Status::IOError(fmt::format("range read for {} returned {} bytes, expected {}", + uri_, bytes_read, fetch_size)); + } + { + std::scoped_lock lock(mutex_); + if (closed_) { + limiter_->Release(fetch_size); + return Status::IOError(fmt::format("{} is closed", uri_)); + } + read_ahead_offset_ = offset; + read_ahead_buffer_ = std::move(fetched); + reserved_read_ahead_size_ = fetch_size; + std::memcpy(buffer, read_ahead_buffer_.data(), static_cast(read_size)); + } + return read_size; + } + + void CloseInternal() { + int64_t reserved = 0; + { + std::scoped_lock lock(mutex_); + closed_ = true; + std::vector().swap(read_ahead_buffer_); + reserved = reserved_read_ahead_size_; + reserved_read_ahead_size_ = 0; + } + limiter_->Release(reserved); + } + + std::shared_ptr client_; + std::shared_ptr limiter_; + ObjectStorePath path_; + std::string uri_; + int64_t length_; + int64_t position_ = 0; + int64_t read_ahead_offset_ = 0; + int64_t next_read_ahead_size_ = kInitialReadAheadSize; + int64_t reserved_read_ahead_size_ = 0; + std::vector read_ahead_buffer_; + bool closed_ = false; + mutable std::mutex read_mutex_; + mutable std::mutex mutex_; +}; + +} // namespace + +ReadAheadMemoryLimiter::ReadAheadMemoryLimiter(int64_t limit) : limit_(limit) {} + +int64_t ReadAheadMemoryLimiter::ReserveUpTo(int64_t min_size, int64_t max_size) { + std::scoped_lock lock(mutex_); + int64_t available = limit_ - used_; + if (available < min_size) { + return 0; + } + int64_t size = std::min(max_size, available); + used_ += size; + return size; +} + +void ReadAheadMemoryLimiter::Release(int64_t size) { + std::scoped_lock lock(mutex_); + assert(size >= 0 && size <= used_); + used_ -= std::clamp(size, int64_t{0}, used_); +} + +ObjectStoreFileSystem::ObjectStoreFileSystem(std::string scheme, + std::shared_ptr client) + : ObjectStoreFileSystem(std::move(scheme), std::move(client), kMaxReadAheadMemory) {} + +ObjectStoreFileSystem::ObjectStoreFileSystem(std::string scheme, + std::shared_ptr client, + int64_t read_ahead_memory_limit) + : scheme_(std::move(scheme)), + client_(std::move(client)), + read_ahead_limiter_(std::make_shared(read_ahead_memory_limit)) {} + +Result ObjectStoreFileSystem::ParsePath(const std::string& path) const { + PAIMON_ASSIGN_OR_RAISE(Path parsed, PathUtil::ToPath(path)); + if (parsed.scheme != scheme_) { + return Status::Invalid(fmt::format("path must use {} scheme: {}", scheme_, path)); + } + if (parsed.authority.empty()) { + return Status::Invalid(fmt::format("{} path must include bucket: {}", scheme_, path)); + } + std::string key = parsed.path; + key.erase(0, key.find_first_not_of('/')); + return ObjectStorePath{parsed.authority, key}; +} + +std::string ObjectStoreFileSystem::ToUri(const ObjectStorePath& path, bool is_dir) const { + std::string uri = scheme_ + "://" + path.bucket + "/"; + uri += path.key; + if (is_dir && uri.back() != '/') { + uri += "/"; + } + return uri; +} + +Result ObjectStoreFileSystem::DirectoryExists(const ObjectStorePath& path) const { + ObjectStorePath directory{path.bucket, NormalizeDirectoryPrefix(path.key)}; + PAIMON_ASSIGN_OR_RAISE(ListObjectsResult result, client_->ListObjects(directory, "", 1)); + return path.key.empty() || !result.objects.empty() || !result.common_prefixes.empty(); +} + +Result> ObjectStoreFileSystem::Open(const std::string& path) const { + PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path)); + if (object_path.key.empty()) { + return Status::Invalid(fmt::format("{} is a directory", path)); + } + Result metadata = client_->HeadObject(object_path); + if (!metadata.ok()) { + if (!metadata.status().IsNotExist()) { + return metadata.status(); + } + PAIMON_ASSIGN_OR_RAISE(bool is_dir, DirectoryExists(object_path)); + if (is_dir) { + return Status::Invalid(fmt::format("{} is a directory", path)); + } + return metadata.status(); + } + return std::make_unique(client_, read_ahead_limiter_, object_path, + ToUri(object_path), metadata.value().size); +} + +Result> ObjectStoreFileSystem::GetFileStatus( + const std::string& path) const { + PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path)); + if (!object_path.key.empty()) { + Result metadata = client_->HeadObject(object_path); + if (metadata.ok()) { + return std::make_unique( + ToUri(object_path), metadata.value().size, metadata.value().modification_time, + false); + } + if (!metadata.status().IsNotExist()) { + return metadata.status(); + } + } + PAIMON_ASSIGN_OR_RAISE(bool exists, DirectoryExists(object_path)); + if (!exists) { + return Status::NotExist(fmt::format("{} does not exist", path)); + } + return std::make_unique(ToUri(object_path, true), 0, 0, true); +} + +Status ObjectStoreFileSystem::ListDirectory( + const ObjectStorePath& path, std::vector>* basic_statuses, + std::vector>* statuses) const { + if (basic_statuses == nullptr && statuses == nullptr) { + return Status::Invalid("a destination status list is required"); + } + ObjectStorePath directory{path.bucket, NormalizeDirectoryPrefix(path.key)}; + std::string token; + do { + PAIMON_ASSIGN_OR_RAISE(ListObjectsResult result, client_->ListObjects(directory, token, 0)); + if (result.is_truncated && result.continuation_token.empty()) { + return Status::IOError( + fmt::format("truncated listing for {} did not include a continuation token", + ToUri(directory, true))); + } + for (const auto& object : result.objects) { + if (object.key == directory.key) { + continue; + } + ObjectStorePath child{path.bucket, object.key}; + if (basic_statuses) { + basic_statuses->push_back( + std::make_unique(ToUri(child), false)); + } else { + statuses->push_back(std::make_unique( + ToUri(child), object.size, object.modification_time, false)); + } + } + for (const auto& prefix : result.common_prefixes) { + ObjectStorePath child{path.bucket, prefix}; + if (basic_statuses) { + basic_statuses->push_back( + std::make_unique(ToUri(child, true), true)); + } else { + statuses->push_back( + std::make_unique(ToUri(child, true), 0, 0, true)); + } + } + token = result.continuation_token; + if (!result.is_truncated) { + break; + } + } while (true); + return Status::OK(); +} + +Status ObjectStoreFileSystem::ListDir( + const std::string& directory, + std::vector>* file_status_list) const { + PAIMON_ASSIGN_OR_RAISE(ObjectStorePath path, ParsePath(directory)); + if (!path.key.empty() && path.key.back() != '/') { + Result metadata = client_->HeadObject(path); + if (metadata.ok()) { + return Status::Invalid(fmt::format("file {} exists and is not a directory", directory)); + } + if (!metadata.status().IsNotExist()) { + return metadata.status(); + } + } + return ListDirectory(path, file_status_list, nullptr); +} + +Status ObjectStoreFileSystem::ListFileStatus( + const std::string& path, std::vector>* file_status_list) const { + PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path)); + if (!object_path.key.empty() && object_path.key.back() != '/') { + Result metadata = client_->HeadObject(object_path); + if (metadata.ok()) { + file_status_list->push_back( + std::make_unique(ToUri(object_path), metadata.value().size, + metadata.value().modification_time, false)); + return Status::OK(); + } + if (!metadata.status().IsNotExist()) { + return metadata.status(); + } + } + return ListDirectory(object_path, nullptr, file_status_list); +} + +Result ObjectStoreFileSystem::Exists(const std::string& path) const { + PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path)); + if (!object_path.key.empty()) { + Result metadata = client_->HeadObject(object_path); + if (metadata.ok()) { + return true; + } + if (!metadata.status().IsNotExist()) { + return metadata.status(); + } + } + return DirectoryExists(object_path); +} + +Status ObjectStoreFileSystem::ReadOnlyStatus() const { + return Status::NotImplemented(fmt::format("{} object store file system is read-only", scheme_)); +} + +Result> ObjectStoreFileSystem::Create(const std::string&, + bool) const { + return ReadOnlyStatus(); +} + +Status ObjectStoreFileSystem::Mkdirs(const std::string&) const { + return ReadOnlyStatus(); +} + +Status ObjectStoreFileSystem::Rename(const std::string&, const std::string&) const { + return ReadOnlyStatus(); +} + +Status ObjectStoreFileSystem::Delete(const std::string&, bool) const { + return ReadOnlyStatus(); +} + +Status ObjectStoreFileSystem::WriteFile(const std::string&, const std::string&, bool) { + return ReadOnlyStatus(); +} + +Status ObjectStoreFileSystem::AtomicStore(const std::string&, const std::string&) { + return ReadOnlyStatus(); +} + +} // namespace paimon::object_store diff --git a/src/paimon/fs/object_store/object_store_file_system.h b/src/paimon/fs/object_store/object_store_file_system.h new file mode 100644 index 000000000..d3698f647 --- /dev/null +++ b/src/paimon/fs/object_store/object_store_file_system.h @@ -0,0 +1,117 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "paimon/fs/file_system.h" +#include "paimon/visibility.h" + +namespace paimon::object_store { + +struct ObjectStorePath { + std::string bucket; + std::string key; +}; + +struct ObjectMetadata { + std::string key; + int64_t size = 0; + int64_t modification_time = 0; +}; + +struct ListObjectsResult { + std::vector objects; + std::vector common_prefixes; + std::string continuation_token; + bool is_truncated = false; +}; + +class PAIMON_EXPORT ObjectStoreClient { + public: + virtual ~ObjectStoreClient() = default; + + virtual Result HeadObject(const ObjectStorePath& path) const = 0; + virtual Result ListObjects(const ObjectStorePath& path, + const std::string& continuation_token, + int max_keys) const = 0; + virtual Result GetObjectRange(const ObjectStorePath& path, int64_t offset, + int64_t size, char* buffer) const = 0; + virtual void GetObjectRangeAsync(const ObjectStorePath& path, int64_t offset, int64_t size, + char* buffer, + std::function&& callback) const = 0; +}; + +class ReadAheadMemoryLimiter { + public: + explicit ReadAheadMemoryLimiter(int64_t limit); + + int64_t ReserveUpTo(int64_t min_size, int64_t max_size); + void Release(int64_t size); + + private: + int64_t limit_; + int64_t used_ = 0; + std::mutex mutex_; +}; + +class PAIMON_EXPORT ObjectStoreFileSystem : public FileSystem { + public: + ObjectStoreFileSystem(std::string scheme, std::shared_ptr client); + ObjectStoreFileSystem(std::string scheme, std::shared_ptr client, + int64_t read_ahead_memory_limit); + ~ObjectStoreFileSystem() override = default; + + Result> Open(const std::string& path) const override; + Result> GetFileStatus(const std::string& path) const override; + Status ListDir(const std::string& directory, + std::vector>* file_status_list) const override; + Status ListFileStatus( + const std::string& path, + std::vector>* file_status_list) const override; + Result Exists(const std::string& path) const override; + + Result> Create(const std::string& path, + bool overwrite) const override; + Status Mkdirs(const std::string& path) const override; + Status Rename(const std::string& src, const std::string& dst) const override; + Status Delete(const std::string& path, bool recursive = true) const override; + Status WriteFile(const std::string& path, const std::string& content, bool overwrite) override; + Status AtomicStore(const std::string& path, const std::string& content) override; + + protected: + Result ParsePath(const std::string& path) const; + std::string ToUri(const ObjectStorePath& path, bool is_dir = false) const; + + private: + Result DirectoryExists(const ObjectStorePath& path) const; + Status ListDirectory(const ObjectStorePath& path, + std::vector>* basic_statuses, + std::vector>* statuses) const; + Status ReadOnlyStatus() const; + + std::string scheme_; + std::shared_ptr client_; + std::shared_ptr read_ahead_limiter_; +}; + +} // namespace paimon::object_store diff --git a/src/paimon/fs/object_store/object_store_file_system_test.cpp b/src/paimon/fs/object_store/object_store_file_system_test.cpp new file mode 100644 index 000000000..8ebea00e0 --- /dev/null +++ b/src/paimon/fs/object_store/object_store_file_system_test.cpp @@ -0,0 +1,273 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/fs/object_store/object_store_file_system.h" + +#include + +#include +#include +#include +#include +#include +#include + +#include "paimon/testing/utils/testharness.h" + +namespace paimon::object_store { +namespace { + +using Range = std::pair; + +class MockObjectStoreClient : public ObjectStoreClient { + public: + Result HeadObject(const ObjectStorePath& path) const override { + ++head_calls_; + if (!head_error_.ok()) { + return head_error_; + } + auto iter = objects_.find(path.key); + if (iter == objects_.end()) { + return Status::NotExist("not found"); + } + return ObjectMetadata{path.key, static_cast(iter->second.size()), 1}; + } + + Result ListObjects(const ObjectStorePath& path, const std::string& token, + int) const override { + ++list_calls_; + if (!list_error_.ok()) { + return list_error_; + } + if (!pages_.empty()) { + return pages_.at(token); + } + ListObjectsResult result; + for (const auto& [key, value] : objects_) { + if (key.rfind(path.key, 0) == 0) { + result.objects.push_back( + ObjectMetadata{key, static_cast(value.size()), 1}); + } + } + return result; + } + + Result GetObjectRange(const ObjectStorePath& path, int64_t offset, int64_t size, + char* buffer) const override { + ranges_.emplace_back(offset, size); + const std::string& value = objects_.at(path.key); + int64_t available = std::min(size, static_cast(value.size()) - offset); + if (short_read_) { + --available; + } + std::memcpy(buffer, value.data() + offset, available); + return available; + } + + void GetObjectRangeAsync(const ObjectStorePath& path, int64_t offset, int64_t size, + char* buffer, std::function&& callback) const override { + Result result = GetObjectRange(path, offset, size, buffer); + callback(result.ok() && result.value() == size ? Status::OK() + : Status::IOError("short read")); + } + + std::map objects_; + std::map pages_; + Status head_error_ = Status::OK(); + Status list_error_ = Status::OK(); + mutable int list_calls_ = 0; + mutable int head_calls_ = 0; + mutable std::vector ranges_; + bool short_read_ = false; +}; + +TEST(ObjectStoreFileSystemTest, TestObjectWinsOverPrefix) { + auto client = std::make_shared(); + client->objects_["foo"] = "file"; + client->objects_["foo/bar"] = "child"; + ObjectStoreFileSystem fs("s3", client); + ASSERT_OK_AND_ASSIGN(auto status, fs.GetFileStatus("s3://bucket/foo")); + ASSERT_FALSE(status->IsDir()); + std::vector> statuses; + ASSERT_OK(fs.ListFileStatus("s3://bucket/foo", &statuses)); + ASSERT_EQ(statuses.size(), 1); + ASSERT_FALSE(statuses[0]->IsDir()); +} + +TEST(ObjectStoreFileSystemTest, TestHeadErrorIsNotMasked) { + auto client = std::make_shared(); + client->head_error_ = Status::IOError("access denied"); + client->objects_["foo/bar"] = "child"; + ObjectStoreFileSystem fs("s3", client); + auto status = fs.GetFileStatus("s3://bucket/foo"); + ASSERT_TRUE(status.status().IsIOError()); + ASSERT_EQ(client->list_calls_, 0); +} + +TEST(ObjectStoreFileSystemTest, TestPaginationAndDirectoryMarker) { + auto client = std::make_shared(); + ListObjectsResult first; + first.objects.push_back({"dir/", 0, 0}); + first.objects.push_back({"dir/a", 1, 1}); + first.is_truncated = true; + first.continuation_token = "next"; + ListObjectsResult second; + second.common_prefixes.push_back("dir/sub/"); + client->pages_[""] = first; + client->pages_["next"] = second; + ObjectStoreFileSystem fs("s3", client); + std::vector> statuses; + ASSERT_OK(fs.ListFileStatus("s3://bucket/dir/", &statuses)); + ASSERT_EQ(statuses.size(), 2); + ASSERT_EQ(statuses[0]->GetPath(), "s3://bucket/dir/a"); + ASSERT_TRUE(statuses[1]->IsDir()); +} + +TEST(ObjectStoreFileSystemTest, TestTruncatedPageRequiresContinuationToken) { + auto client = std::make_shared(); + ListObjectsResult page; + page.objects.push_back({"dir/a", 1, 1}); + page.is_truncated = true; + client->pages_[""] = page; + ObjectStoreFileSystem fs("s3", client); + std::vector> statuses; + ASSERT_TRUE(fs.ListFileStatus("s3://bucket/dir/", &statuses).IsIOError()); + ASSERT_TRUE(statuses.empty()); +} + +TEST(ObjectStoreFileSystemTest, TestOpenBucketRootIsDirectory) { + auto client = std::make_shared(); + ObjectStoreFileSystem fs("s3", client); + ASSERT_TRUE(fs.Open("s3://bucket/").status().IsInvalid()); + ASSERT_EQ(client->head_calls_, 0); + ASSERT_EQ(client->list_calls_, 0); +} + +TEST(ObjectStoreFileSystemTest, TestPathWithLeadingSlashes) { + auto client = std::make_shared(); + client->objects_["file"] = "data"; + ObjectStoreFileSystem fs("s3", client); + ASSERT_OK_AND_ASSIGN(auto status, fs.GetFileStatus("s3://bucket///file")); + ASSERT_EQ(status->GetPath(), "s3://bucket/file"); +} + +TEST(ObjectStoreInputStreamTest, TestBoundsCloseAndSeekOverflow) { + auto client = std::make_shared(); + client->objects_["file"] = std::string(128 * 1024, 'x'); + ObjectStoreFileSystem fs("s3", client, 64 * 1024); + ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file")); + char data[4]; + bool called = false; + stream->ReadAsync(data, 4, 128 * 1024 - 2, [&called](Status status) { + called = true; + ASSERT_TRUE(status.IsInvalid()); + }); + ASSERT_TRUE(called); + ASSERT_TRUE(stream->Seek(std::numeric_limits::max(), FS_SEEK_END).IsInvalid()); + ASSERT_OK(stream->Close()); + called = false; + stream->ReadAsync(data, 1, 0, [&called, &stream](Status status) { + called = true; + ASSERT_TRUE(status.IsIOError()); + ASSERT_TRUE(stream->GetPos().status().IsIOError()); + }); + ASSERT_TRUE(called); +} + +TEST(ObjectStoreInputStreamTest, TestShortReadFails) { + auto client = std::make_shared(); + client->objects_["file"] = std::string(128 * 1024, 'x'); + client->short_read_ = true; + ObjectStoreFileSystem fs("s3", client); + ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file")); + char data[4]; + ASSERT_TRUE(stream->Read(data, sizeof(data)).status().IsIOError()); +} + +TEST(ObjectStoreInputStreamTest, TestSequentialReadAheadGrows) { + auto client = std::make_shared(); + client->objects_["file"] = std::string(1024 * 1024, 'x'); + ObjectStoreFileSystem fs("s3", client); + ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file")); + std::array buffer{}; + ASSERT_OK(stream->Read(buffer.data(), buffer.size())); + ASSERT_EQ(client->ranges_.back(), Range(0, 64 * 1024)); + ASSERT_OK(stream->Seek(64 * 1024, FS_SEEK_SET)); + ASSERT_OK(stream->Read(buffer.data(), buffer.size())); + ASSERT_EQ(client->ranges_.back(), Range(64 * 1024, 128 * 1024)); +} + +TEST(ObjectStoreInputStreamTest, TestConsumedBufferReleasesBudget) { + auto client = std::make_shared(); + client->objects_["file"] = std::string(1024 * 1024, 'x'); + ObjectStoreFileSystem fs("s3", client, 64 * 1024); + ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file")); + std::array buffer{}; + for (int i = 0; i < 17; ++i) { + ASSERT_OK(stream->Read(buffer.data(), buffer.size())); + } + ASSERT_EQ(client->ranges_.size(), 2); + ASSERT_EQ(client->ranges_.back(), Range(64 * 1024, 64 * 1024)); +} + +TEST(ObjectStoreInputStreamTest, TestBackwardSeekReleasesBudget) { + auto client = std::make_shared(); + client->objects_["file"] = std::string(1024 * 1024, 'x'); + ObjectStoreFileSystem fs("s3", client, 64 * 1024); + ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file")); + std::array buffer{}; + ASSERT_OK(stream->Seek(64 * 1024, FS_SEEK_SET)); + ASSERT_OK(stream->Read(buffer.data(), buffer.size())); + ASSERT_OK(stream->Seek(0, FS_SEEK_SET)); + ASSERT_OK(stream->Read(buffer.data(), buffer.size())); + ASSERT_EQ(client->ranges_.back(), Range(0, 64 * 1024)); +} + +TEST(ObjectStoreInputStreamTest, TestLargeDirectReadReleasesBudget) { + auto client = std::make_shared(); + client->objects_["file"] = std::string(16 * 1024 * 1024, 'x'); + ObjectStoreFileSystem fs("s3", client, 64 * 1024); + ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file")); + std::array small_buffer{}; + ASSERT_OK(stream->Seek(64 * 1024, FS_SEEK_SET)); + ASSERT_OK(stream->Read(small_buffer.data(), small_buffer.size())); + std::vector large_buffer(8 * 1024 * 1024); + ASSERT_OK(stream->Seek(1024 * 1024, FS_SEEK_SET)); + ASSERT_OK(stream->Read(large_buffer.data(), large_buffer.size())); + ASSERT_OK(stream->Seek(0, FS_SEEK_SET)); + ASSERT_OK(stream->Read(small_buffer.data(), small_buffer.size())); + ASSERT_EQ(client->ranges_.back(), Range(0, 64 * 1024)); +} + +TEST(ObjectStoreInputStreamTest, TestCompetingStreamsShareBudget) { + auto client = std::make_shared(); + client->objects_["first"] = std::string(1024 * 1024, 'x'); + client->objects_["second"] = std::string(1024 * 1024, 'x'); + ObjectStoreFileSystem fs("s3", client, 64 * 1024); + ASSERT_OK_AND_ASSIGN(auto first, fs.Open("s3://bucket/first")); + ASSERT_OK_AND_ASSIGN(auto second, fs.Open("s3://bucket/second")); + std::array buffer{}; + ASSERT_OK(first->Read(buffer.data(), buffer.size())); + ASSERT_OK(second->Read(buffer.data(), buffer.size())); + ASSERT_EQ(client->ranges_.back(), Range(0, 4096)); + ASSERT_OK(first->Close()); + ASSERT_OK(second->Seek(100000, FS_SEEK_SET)); + ASSERT_OK(second->Read(buffer.data(), buffer.size())); + ASSERT_EQ(client->ranges_.back(), Range(100000, 64 * 1024)); +} + +} // namespace +} // namespace paimon::object_store diff --git a/src/paimon/fs/s3/CMakeLists.txt b/src/paimon/fs/s3/CMakeLists.txt new file mode 100644 index 000000000..11c608f6d --- /dev/null +++ b/src/paimon/fs/s3/CMakeLists.txt @@ -0,0 +1,55 @@ +# Copyright 2026-present Alibaba Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +if(PAIMON_ENABLE_S3) + add_paimon_lib(paimon_s3_file_system + SOURCES + s3_file_system.cpp + s3_file_system_factory.cpp + EXTRA_INCLUDES + ${AWS_AUTH_INCLUDE_DIR} + DEPENDENCIES + paimon_shared + CURL::libcurl + STATIC_LINK_LIBS + CURL::libcurl + aws_auth_minimal + fmt + SHARED_LINK_LIBS + paimon_shared + paimon_object_store_shared + SHARED_LINK_FLAGS + ${PAIMON_VERSION_SCRIPT_FLAGS}) + + add_dependencies(paimon_s3_file_system_objlib aws-c-auth_ep) + + if(TARGET paimon_s3_file_system_static) + target_link_libraries(paimon_s3_file_system_static + PRIVATE paimon_object_store_static) + endif() + + if(PAIMON_BUILD_TESTS) + add_paimon_test(s3_file_system_test + SOURCES + s3_file_system_test.cpp + EXTRA_INCLUDES + ${AWS_AUTH_INCLUDE_DIR} + STATIC_LINK_LIBS + paimon_shared + test_utils_static + ${PAIMON_LOCAL_FILE_SYSTEM_STATIC_LINK_LIBS} + ${PAIMON_S3_FILE_SYSTEM_STATIC_LINK_LIBS} + ${GTEST_LINK_TOOLCHAIN}) + endif() +endif() diff --git a/src/paimon/fs/s3/s3_file_system.cpp b/src/paimon/fs/s3/s3_file_system.cpp new file mode 100644 index 000000000..e0eda235a --- /dev/null +++ b/src/paimon/fs/s3/s3_file_system.cpp @@ -0,0 +1,771 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/fs/s3/s3_file_system.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "fmt/format.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/executor.h" +#include "paimon/fs/object_store/http_client.h" + +namespace paimon::s3 { +namespace { + +using object_store::HttpBodyConsumer; +using object_store::HttpClient; +using object_store::HttpHeaders; +using object_store::HttpMethod; +using object_store::HttpRequest; +using object_store::HttpResponse; +using object_store::ListObjectsResult; +using object_store::ObjectMetadata; +using object_store::ObjectStoreClient; +using object_store::ObjectStorePath; + +Result PercentEncode(std::string_view value, bool preserve_slash) { + if (value.size() > std::numeric_limits::max()) { + return Status::IOError("S3 URL component is too large to encode"); + } + char* encoded = curl_easy_escape(nullptr, value.data(), static_cast(value.size())); + if (encoded == nullptr) { + return Status::IOError("failed to URL encode S3 component"); + } + std::string result(encoded); + curl_free(encoded); + if (preserve_slash) { + result = StringUtils::Replace(result, "%2F", "/"); + } + return result; +} + +Result PercentDecode(std::string_view value, const std::string& field) { + for (size_t position = 0; position < value.size(); ++position) { + if (value[position] == '%' && + (position + 2 >= value.size() || + !std::isxdigit(static_cast(value[position + 1])) || + !std::isxdigit(static_cast(value[position + 2])))) { + return Status::IOError(fmt::format("invalid URL encoding in S3 {}", field)); + } + } + if (value.size() > std::numeric_limits::max()) { + return Status::IOError(fmt::format("S3 {} is too large to URL decode", field)); + } + int decoded_size = 0; + char* decoded = + curl_easy_unescape(nullptr, value.data(), static_cast(value.size()), &decoded_size); + if (decoded == nullptr) { + return Status::IOError(fmt::format("failed to URL decode S3 {}", field)); + } + std::string result(decoded, decoded_size); + curl_free(decoded); + return result; +} + +Result ParseNonNegativeInt64(const std::string& value, const std::string& field) { + std::optional result = StringUtils::StringToValue(value); + if (!result || *result < 0 || (!value.empty() && value.front() == '-')) { + return Status::IOError(fmt::format("S3 {} is not a non-negative integer", field)); + } + 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[] = { + {"&", "&"}, {"<", "<"}, {">", ">"}, {""", "\""}, {"'", "'"}}; + + std::string result; + result.reserve(value.size()); + for (size_t position = 0; position < value.size();) { + bool matched = false; + if (value[position] == '&') { + for (const auto& [entity, replacement] : entities) { + size_t entity_size = std::strlen(entity); + if (value.compare(position, entity_size, entity) == 0) { + result.append(replacement); + position += entity_size; + matched = true; + break; + } + } + } + if (!matched) { + result.push_back(value[position++]); + } + } + return result; +} + +std::optional TagValue(const std::string& xml, const std::string& tag, + size_t offset = 0) { + std::string begin = "<" + tag + ">"; + std::string end = ""; + size_t begin_position = xml.find(begin, offset); + if (begin_position == std::string::npos) { + return std::nullopt; + } + begin_position += begin.size(); + size_t end_position = xml.find(end, begin_position); + if (end_position == std::string::npos) { + return std::nullopt; + } + return XmlUnescape(xml.substr(begin_position, end_position - begin_position)); +} + +Result> TagBlocks(const std::string& xml, const std::string& tag) { + std::vector blocks; + std::string begin = "<" + tag + ">"; + std::string end = ""; + size_t position = 0; + while (true) { + size_t begin_position = xml.find(begin, position); + size_t unexpected_end = xml.find(end, position); + if (begin_position == std::string::npos) { + if (unexpected_end != std::string::npos) { + return Status::IOError(fmt::format("malformed S3 XML element {}", tag)); + } + break; + } + if (unexpected_end != std::string::npos && unexpected_end < begin_position) { + return Status::IOError(fmt::format("malformed S3 XML element {}", tag)); + } + size_t end_position = xml.find(end, begin_position + begin.size()); + if (end_position == std::string::npos) { + return Status::IOError(fmt::format("malformed S3 XML element {}", tag)); + } + size_t nested_begin = xml.find(begin, begin_position + begin.size()); + if (nested_begin != std::string::npos && nested_begin < end_position) { + return Status::IOError(fmt::format("malformed S3 XML element {}", tag)); + } + end_position += end.size(); + blocks.push_back(xml.substr(begin_position, end_position - begin_position)); + position = end_position; + } + return blocks; +} + +class AwsAuthRuntime { + public: + AwsAuthRuntime() : allocator_(aws_default_allocator()) { + aws_auth_library_init(allocator_); + event_loop_group_ = aws_event_loop_group_new_default(allocator_, 1, nullptr); + aws_host_resolver_default_options resolver_options{}; + resolver_options.el_group = event_loop_group_; + resolver_options.max_entries = 8; + resolver_ = aws_host_resolver_new_default(allocator_, &resolver_options); + aws_client_bootstrap_options bootstrap_options{}; + bootstrap_options.event_loop_group = event_loop_group_; + bootstrap_options.host_resolver = resolver_; + bootstrap_ = aws_client_bootstrap_new(allocator_, &bootstrap_options); + aws_tls_ctx_options tls_options; + aws_tls_ctx_options_init_default_client(&tls_options, allocator_); + tls_context_ = aws_tls_client_ctx_new(allocator_, &tls_options); + aws_tls_ctx_options_clean_up(&tls_options); + } + + aws_allocator* allocator() const { + return allocator_; + } + aws_client_bootstrap* bootstrap() const { + return bootstrap_; + } + aws_tls_ctx* tls_context() const { + return tls_context_; + } + + private: + aws_allocator* allocator_; + aws_event_loop_group* event_loop_group_; + aws_host_resolver* resolver_; + aws_client_bootstrap* bootstrap_; + aws_tls_ctx* tls_context_; +}; + +AwsAuthRuntime& GetAwsAuthRuntime() { + static auto* runtime = new AwsAuthRuntime(); + return *runtime; +} + +aws_byte_cursor Cursor(const std::string& value) { + return aws_byte_cursor_from_array(value.data(), value.size()); +} + +std::shared_ptr WrapProvider(aws_credentials_provider* provider) { + return std::shared_ptr(provider, aws_credentials_provider_release); +} + +std::string ResolveRegion(const std::map& options) { + auto region = options.find(kS3RegionOption); + if (region != options.end() && !region->second.empty()) { + return region->second; + } + const char* environment_region = std::getenv("AWS_REGION"); + if (environment_region != nullptr && environment_region[0] != '\0') { + return environment_region; + } + environment_region = std::getenv("AWS_DEFAULT_REGION"); + if (environment_region != nullptr && environment_region[0] != '\0') { + return environment_region; + } + + AwsAuthRuntime& runtime = GetAwsAuthRuntime(); + aws_byte_cursor profile_override{}; + const aws_byte_cursor* profile_override_ptr = nullptr; + auto profile = options.find(kS3ProfileOption); + if (profile != options.end() && !profile->second.empty()) { + profile_override = Cursor(profile->second); + profile_override_ptr = &profile_override; + } + aws_string* config_path = aws_get_config_file_path(runtime.allocator(), nullptr); + aws_string* profile_name = aws_get_profile_name(runtime.allocator(), profile_override_ptr); + aws_profile_collection* profiles = config_path == nullptr + ? nullptr + : aws_profile_collection_new_from_file( + runtime.allocator(), config_path, AWS_PST_CONFIG); + const aws_profile* selected_profile = + profiles == nullptr || profile_name == nullptr + ? nullptr + : aws_profile_collection_get_profile(profiles, profile_name); + aws_string* region_name = aws_string_new_from_c_str(runtime.allocator(), "region"); + const aws_profile_property* property = + selected_profile == nullptr || region_name == nullptr + ? nullptr + : aws_profile_get_property(selected_profile, region_name); + const aws_string* value = + property == nullptr ? nullptr : aws_profile_property_get_value(property); + std::string resolved = value == nullptr ? "" : aws_string_c_str(value); + aws_string_destroy(region_name); + aws_profile_collection_release(profiles); + aws_string_destroy(profile_name); + aws_string_destroy(config_path); + return resolved.empty() ? "us-east-1" : resolved; +} + +std::shared_ptr MakeCredentialsProvider( + const std::map& options) { + AwsAuthRuntime& runtime = GetAwsAuthRuntime(); + auto access = options.find(kS3AccessKeyIdOption); + if (access != options.end()) { + const std::string& secret = options.at(kS3SecretAccessKeyOption); + std::string token; + auto token_iter = options.find(kS3SessionTokenOption); + if (token_iter != options.end()) { + token = token_iter->second; + } + aws_credentials_provider_static_options static_options{}; + static_options.access_key_id = Cursor(access->second); + static_options.secret_access_key = Cursor(secret); + static_options.session_token = Cursor(token); + return WrapProvider( + aws_credentials_provider_new_static(runtime.allocator(), &static_options)); + } + + aws_byte_cursor profile_override{}; + auto profile_iter = options.find(kS3ProfileOption); + if (profile_iter != options.end() && !profile_iter->second.empty()) { + profile_override = Cursor(profile_iter->second); + } + std::string region; + auto region_iter = options.find(kS3RegionOption); + if (region_iter != options.end()) { + region = region_iter->second; + } + + std::vector providers; + aws_credentials_provider_environment_options environment_options{}; + providers.push_back( + aws_credentials_provider_new_environment(runtime.allocator(), &environment_options)); + + aws_credentials_provider_profile_options profile_options{}; + profile_options.profile_name_override = profile_override; + profile_options.bootstrap = runtime.bootstrap(); + profile_options.tls_ctx = runtime.tls_context(); + providers.push_back( + aws_credentials_provider_new_profile(runtime.allocator(), &profile_options)); + + aws_credentials_provider_sts_web_identity_options web_options{}; + web_options.profile_name_override = profile_override; + web_options.region = Cursor(region); + web_options.bootstrap = runtime.bootstrap(); + web_options.tls_ctx = runtime.tls_context(); + providers.push_back( + aws_credentials_provider_new_sts_web_identity(runtime.allocator(), &web_options)); + + aws_credentials_provider_sso_options sso_options{}; + sso_options.profile_name_override = profile_override; + sso_options.bootstrap = runtime.bootstrap(); + sso_options.tls_ctx = runtime.tls_context(); + providers.push_back(aws_credentials_provider_new_sso(runtime.allocator(), &sso_options)); + + aws_credentials_provider_login_options login_options{}; + login_options.profile_name_override = profile_override; + login_options.login_region = Cursor(region); + login_options.bootstrap = runtime.bootstrap(); + login_options.tls_ctx = runtime.tls_context(); + providers.push_back(aws_credentials_provider_new_login(runtime.allocator(), &login_options)); + + aws_credentials_provider_chain_default_options default_options{}; + default_options.profile_name_override = profile_override; + default_options.bootstrap = runtime.bootstrap(); + default_options.tls_ctx = runtime.tls_context(); + default_options.skip_environment_credentials_provider = true; + providers.push_back( + aws_credentials_provider_new_chain_default(runtime.allocator(), &default_options)); + + providers.erase(std::remove(providers.begin(), providers.end(), nullptr), providers.end()); + aws_credentials_provider_chain_options chain_options{}; + chain_options.providers = providers.data(); + chain_options.provider_count = providers.size(); + aws_credentials_provider* chain = + aws_credentials_provider_new_chain(runtime.allocator(), &chain_options); + for (aws_credentials_provider* provider : providers) { + aws_credentials_provider_release(provider); + } + if (chain == nullptr) { + return nullptr; + } + aws_credentials_provider_cached_options cached_options{}; + cached_options.source = chain; + cached_options.refresh_time_in_milliseconds = 15 * 60 * 1000; + aws_credentials_provider* cached = + aws_credentials_provider_new_cached(runtime.allocator(), &cached_options); + aws_credentials_provider_release(chain); + return WrapProvider(cached); +} + +struct Endpoint { + std::string scheme; + std::string authority; + std::string base_path; +}; + +Endpoint ParseEndpoint(std::string endpoint) { + if (endpoint.find("://") == std::string::npos) { + endpoint = "https://" + endpoint; + } + size_t scheme_end = endpoint.find("://"); + size_t path_begin = endpoint.find('/', scheme_end + 3); + return Endpoint{endpoint.substr(0, scheme_end), + endpoint.substr(scheme_end + 3, path_begin - scheme_end - 3), + path_begin == std::string::npos ? "" : endpoint.substr(path_begin)}; +} + +struct SigningContext { + SigningContext(aws_allocator* allocator, aws_http_message* message) + : allocator(allocator), message(message) {} + + aws_allocator* allocator; + aws_http_message* message; + std::mutex mutex; + std::condition_variable condition; + int error_code = AWS_ERROR_SUCCESS; + bool complete = false; +}; + +void OnSigningComplete(aws_signing_result* result, int error_code, void* user_data) { + auto* context = static_cast(user_data); + if (error_code == AWS_ERROR_SUCCESS && + aws_apply_signing_result_to_http_request(context->message, context->allocator, result)) { + error_code = aws_last_error(); + } + { + std::scoped_lock lock(context->mutex); + context->error_code = error_code; + context->complete = true; + } + context->condition.notify_one(); +} + +class S3ObjectStoreClient : public ObjectStoreClient, + public std::enable_shared_from_this { + public: + S3ObjectStoreClient(const std::map& options, + std::shared_ptr http_client, + std::shared_ptr credentials) + : http_client_(std::move(http_client)), credentials_(std::move(credentials)) { + region_ = ResolveRegion(options); + auto endpoint = options.find(kS3EndpointOption); + use_default_endpoint_ = endpoint == options.end() || endpoint->second.empty(); + endpoint_ = ParseEndpoint(use_default_endpoint_ + ? fmt::format("https://s3.{}.amazonaws.com", region_) + : endpoint->second); + auto path_style = options.find(kS3PathStyleAccessOption); + path_style_configured_ = path_style != options.end(); + path_style_ = path_style_configured_ && ParseBoolOption(path_style->second).value(); + } + + Result HeadObject(const ObjectStorePath& path) const override { + PAIMON_ASSIGN_OR_RAISE(HttpResponse response, + Execute(path, HttpMethod::HEAD, "", {}, nullptr)); + if (response.status_code == 404) { + return Status::NotExist( + fmt::format("s3://{}/{} does not exist", path.bucket, path.key)); + } + PAIMON_RETURN_NOT_OK(CheckResponse(response, "HeadObject", path)); + auto length = response.headers.find("content-length"); + if (length == response.headers.end()) { + return Status::IOError("HeadObject response is missing Content-Length"); + } + int64_t modification_time = 0; + auto modified = response.headers.find("last-modified"); + if (modified != response.headers.end()) { + modification_time = ParseModificationTime(modified->second); + } + PAIMON_ASSIGN_OR_RAISE(int64_t object_size, + ParseNonNegativeInt64(length->second, "Content-Length")); + return ObjectMetadata{path.key, object_size, modification_time}; + } + + Result ListObjects(const ObjectStorePath& path, + const std::string& continuation_token, + int max_keys) const override { + std::string query = "list-type=2&delimiter=%2F&encoding-type=url"; + if (!path.key.empty()) { + PAIMON_ASSIGN_OR_RAISE(std::string encoded_prefix, PercentEncode(path.key, false)); + query += "&prefix=" + encoded_prefix; + } + if (!continuation_token.empty()) { + PAIMON_ASSIGN_OR_RAISE(std::string encoded_token, + PercentEncode(continuation_token, false)); + query += "&continuation-token=" + encoded_token; + } + if (max_keys > 0) { + query += "&max-keys=" + std::to_string(max_keys); + } + std::string body; + HttpBodyConsumer consumer = [&body](const char* data, int64_t size) { + body.append(data, static_cast(size)); + return Status::OK(); + }; + ObjectStorePath bucket_path{path.bucket, ""}; + PAIMON_ASSIGN_OR_RAISE(HttpResponse response, + Execute(bucket_path, HttpMethod::GET, query, {}, consumer)); + if (response.status_code == 404) { + return Status::NotExist(fmt::format("S3 bucket {} does not exist", path.bucket)); + } + PAIMON_RETURN_NOT_OK(CheckResponse(response, "ListObjectsV2", path)); + if (body.find("") == std::string::npos) { + return Status::IOError("malformed S3 ListObjectsV2 XML response"); + } + ListObjectsResult result; + PAIMON_ASSIGN_OR_RAISE(std::vector contents, TagBlocks(body, "Contents")); + for (const std::string& block : contents) { + auto key = TagValue(block, "Key"); + auto size = TagValue(block, "Size"); + if (!key || !size) { + return Status::IOError("S3 ListObjectsV2 Contents is missing Key or Size"); + } + 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; + auto last_modified = TagValue(block, "LastModified"); + if (last_modified) { + modified = ParseModificationTime(*last_modified); + } + result.objects.push_back(ObjectMetadata{decoded_key, object_size, modified}); + } + PAIMON_ASSIGN_OR_RAISE(std::vector common_prefixes, + TagBlocks(body, "CommonPrefixes")); + for (const std::string& block : common_prefixes) { + auto prefix = TagValue(block, "Prefix"); + if (!prefix) { + return Status::IOError("S3 ListObjectsV2 CommonPrefixes is missing Prefix"); + } + PAIMON_ASSIGN_OR_RAISE(std::string decoded_prefix, PercentDecode(*prefix, "Prefix")); + result.common_prefixes.push_back(std::move(decoded_prefix)); + } + auto is_truncated = TagValue(body, "IsTruncated"); + if (!is_truncated || (*is_truncated != "true" && *is_truncated != "false")) { + return Status::IOError("S3 ListObjectsV2 response has invalid IsTruncated"); + } + result.is_truncated = *is_truncated == "true"; + auto token = TagValue(body, "NextContinuationToken"); + if (token) { + PAIMON_ASSIGN_OR_RAISE(result.continuation_token, + PercentDecode(*token, "NextContinuationToken")); + } + return result; + } + + Result GetObjectRange(const ObjectStorePath& path, int64_t offset, int64_t size, + char* buffer) const override { + if (size == 0) { + return 0; + } + int64_t copied = 0; + HttpHeaders headers{{"range", fmt::format("bytes={}-{}", offset, offset + size - 1)}}; + HttpBodyConsumer consumer = [&copied, buffer, size](const char* data, int64_t length) { + if (length > size - copied) { + return Status::IOError("S3 range response exceeds the requested length"); + } + std::memcpy(buffer + copied, data, static_cast(length)); + copied += length; + return Status::OK(); + }; + PAIMON_ASSIGN_OR_RAISE(HttpResponse response, + Execute(path, HttpMethod::GET, "", headers, consumer)); + if (response.status_code == 404) { + return Status::NotExist( + fmt::format("s3://{}/{} does not exist", path.bucket, path.key)); + } + PAIMON_RETURN_NOT_OK(CheckResponse(response, "GetObject", path)); + if (copied != size) { + return Status::IOError( + fmt::format("GetObject read {} bytes for s3://{}/{}, expected {}", copied, + path.bucket, path.key, size)); + } + return copied; + } + + void GetObjectRangeAsync(const ObjectStorePath& path, int64_t offset, int64_t size, + char* buffer, std::function&& callback) const override { + auto self = shared_from_this(); + GetGlobalDefaultExecutor()->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 CheckResponse(const HttpResponse& response, const std::string& operation, + const ObjectStorePath& path) const { + if (response.status_code >= 200 && response.status_code < 300) { + return Status::OK(); + } + return Status::IOError(fmt::format("{} failed for s3://{}/{}: HTTP {}", operation, + path.bucket, path.key, response.status_code)); + } + + Result Execute(const ObjectStorePath& object, HttpMethod method, + const std::string& query, const HttpHeaders& headers, + const HttpBodyConsumer& consumer) const { + std::string authority = endpoint_.authority; + std::string request_path = endpoint_.base_path; + if (request_path.empty() || request_path.back() != '/') { + request_path += '/'; + } + bool use_path_style = path_style_ || (!path_style_configured_ && use_default_endpoint_ && + endpoint_.scheme == "https" && + object.bucket.find('.') != std::string::npos); + if (use_path_style) { + PAIMON_ASSIGN_OR_RAISE(std::string encoded_bucket, PercentEncode(object.bucket, false)); + request_path += encoded_bucket + "/"; + } else { + authority = object.bucket + "." + authority; + } + PAIMON_ASSIGN_OR_RAISE(std::string encoded_key, PercentEncode(object.key, true)); + request_path += encoded_key; + if (!query.empty()) { + request_path += "?" + query; + } + + AwsAuthRuntime& runtime = GetAwsAuthRuntime(); + aws_http_message* message = aws_http_message_new_request(runtime.allocator()); + std::string method_name = method == HttpMethod::HEAD ? "HEAD" : "GET"; + aws_http_message_set_request_method(message, Cursor(method_name)); + aws_http_message_set_request_path(message, Cursor(request_path)); + aws_http_header host_header{}; + host_header.name = aws_byte_cursor_from_c_str("host"); + host_header.value = Cursor(authority); + aws_http_message_add_header(message, host_header); + for (const auto& [name, value] : headers) { + aws_http_header header{}; + header.name = Cursor(name); + header.value = Cursor(value); + aws_http_message_add_header(message, header); + } + aws_signable* signable = aws_signable_new_http_request(runtime.allocator(), message); + aws_signing_config_aws config{}; + config.config_type = AWS_SIGNING_CONFIG_AWS; + config.algorithm = AWS_SIGNING_ALGORITHM_V4; + config.signature_type = AWS_ST_HTTP_REQUEST_HEADERS; + config.region = Cursor(region_); + config.service = aws_byte_cursor_from_c_str("s3"); + aws_date_time_init_now(&config.date); + config.flags.use_double_uri_encode = false; + config.flags.should_normalize_uri_path = false; + config.signed_body_value = g_aws_signed_body_value_unsigned_payload; + config.signed_body_header = AWS_SBHT_X_AMZ_CONTENT_SHA256; + config.credentials_provider = credentials_.get(); + + SigningContext context(runtime.allocator(), message); + int result = aws_sign_request_aws(runtime.allocator(), signable, + reinterpret_cast(&config), + OnSigningComplete, &context); + if (result == AWS_OP_SUCCESS) { + std::unique_lock lock(context.mutex); + context.condition.wait(lock, [&context] { return context.complete; }); + } + aws_signable_destroy(signable); + if (result != AWS_OP_SUCCESS || context.error_code != AWS_ERROR_SUCCESS) { + int error = result == AWS_OP_SUCCESS ? context.error_code : aws_last_error(); + aws_http_message_release(message); + return Status::IOError( + fmt::format("failed to sign S3 request: {}", aws_error_debug_str(error))); + } + + HttpRequest request; + request.method = method; + request.url = endpoint_.scheme + "://" + authority + request_path; + aws_http_headers* signed_headers = aws_http_message_get_headers(message); + for (size_t i = 0; i < aws_http_headers_count(signed_headers); ++i) { + aws_http_header header; + aws_http_headers_get_index(signed_headers, i, &header); + request.headers[std::string(reinterpret_cast(header.name.ptr), + header.name.len)] = + std::string(reinterpret_cast(header.value.ptr), header.value.len); + } + aws_http_message_release(message); + HttpBodyConsumer body_consumer = consumer; + if (!body_consumer) { + body_consumer = [](const char*, int64_t) { return Status::OK(); }; + } + return http_client_->Execute(request, body_consumer); + } + + std::shared_ptr http_client_; + std::shared_ptr credentials_; + Endpoint endpoint_; + std::string region_; + bool path_style_ = false; + bool path_style_configured_ = false; + bool use_default_endpoint_ = false; +}; + +class InvalidObjectStoreClient : public ObjectStoreClient { + public: + explicit InvalidObjectStoreClient(Status status) : status_(std::move(status)) {} + + Result HeadObject(const ObjectStorePath&) const override { + return status_; + } + + Result ListObjects(const ObjectStorePath&, const std::string&, + int) const override { + return status_; + } + + Result GetObjectRange(const ObjectStorePath&, int64_t, int64_t, char*) const override { + return status_; + } + + void GetObjectRangeAsync(const ObjectStorePath&, int64_t, int64_t, char*, + std::function&& callback) const override { + callback(status_); + } + + private: + Status status_; +}; + +} // namespace + +Result ParseBoolOption(const std::string& value) { + std::string lower = StringUtils::ToLowerCase(value); + if (lower == "true" || lower == "1" || lower == "yes" || lower == "on") { + return true; + } + if (lower == "false" || lower == "0" || lower == "no" || lower == "off") { + return false; + } + return Status::Invalid(fmt::format("invalid boolean value: {}", value)); +} + +Status ValidateS3Options(const std::map& options) { + auto access = options.find(kS3AccessKeyIdOption); + auto secret = options.find(kS3SecretAccessKeyOption); + bool has_access = access != options.end(); + bool has_secret = secret != options.end(); + bool has_token = options.find(kS3SessionTokenOption) != options.end(); + if (has_access != has_secret) { + return Status::Invalid(fmt::format("{} and {} must be configured together", + kS3AccessKeyIdOption, kS3SecretAccessKeyOption)); + } + if (has_token && !has_access) { + return Status::Invalid(fmt::format("{} requires {} and {}", kS3SessionTokenOption, + kS3AccessKeyIdOption, kS3SecretAccessKeyOption)); + } + if (has_access && access->second.empty()) { + return Status::Invalid(fmt::format("{} must not be empty", kS3AccessKeyIdOption)); + } + if (has_secret && secret->second.empty()) { + return Status::Invalid(fmt::format("{} must not be empty", kS3SecretAccessKeyOption)); + } + auto path_style = options.find(kS3PathStyleAccessOption); + if (path_style == options.end()) { + return Status::OK(); + } + Result parsed = ParseBoolOption(path_style->second); + if (!parsed.ok()) { + return Status::Invalid( + fmt::format("{} {}", kS3PathStyleAccessOption, parsed.status().message())); + } + return Status::OK(); +} + +S3FileSystem::S3FileSystem(const std::map& options) + : ObjectStoreFileSystem("s3", MakeS3ObjectStoreClient( + options, std::make_shared())) {} + +std::shared_ptr MakeS3ObjectStoreClient( + const std::map& options, + std::shared_ptr http_client) { + Status status = ValidateS3Options(options); + if (!status.ok()) { + return std::make_shared(std::move(status)); + } + auto credentials = MakeCredentialsProvider(options); + if (!credentials) { + return std::make_shared( + Status::IOError("failed to initialize S3 credentials provider")); + } + return std::make_shared(options, std::move(http_client), + std::move(credentials)); +} + +} // namespace paimon::s3 diff --git a/src/paimon/fs/s3/s3_file_system.h b/src/paimon/fs/s3/s3_file_system.h new file mode 100644 index 000000000..4d712d56e --- /dev/null +++ b/src/paimon/fs/s3/s3_file_system.h @@ -0,0 +1,48 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include + +#include "paimon/fs/object_store/http_client.h" +#include "paimon/fs/object_store/object_store_file_system.h" + +namespace paimon::s3 { + +inline constexpr char kS3RegionOption[] = "s3.region"; +inline constexpr char kS3EndpointOption[] = "s3.endpoint"; +inline constexpr char kS3PathStyleAccessOption[] = "s3.path-style-access"; +inline constexpr char kS3ProfileOption[] = "s3.profile"; +inline constexpr char kS3AccessKeyIdOption[] = "s3.access-key-id"; +inline constexpr char kS3SecretAccessKeyOption[] = "s3.secret-access-key"; +inline constexpr char kS3SessionTokenOption[] = "s3.session-token"; + +Result ParseBoolOption(const std::string& value); +Status ValidateS3Options(const std::map& options); +std::shared_ptr MakeS3ObjectStoreClient( + const std::map& options, + std::shared_ptr http_client); + +class S3FileSystem : public object_store::ObjectStoreFileSystem { + public: + explicit S3FileSystem(const std::map& options); + ~S3FileSystem() override = default; +}; + +} // namespace paimon::s3 diff --git a/src/paimon/fs/s3/s3_file_system_factory.cpp b/src/paimon/fs/s3/s3_file_system_factory.cpp new file mode 100644 index 000000000..80bd8e67a --- /dev/null +++ b/src/paimon/fs/s3/s3_file_system_factory.cpp @@ -0,0 +1,34 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/fs/s3/s3_file_system_factory.h" + +#include "paimon/factories/factory.h" +#include "paimon/fs/s3/s3_file_system.h" + +namespace paimon::s3 { + +const char S3FileSystemFactory::IDENTIFIER[] = "s3"; + +Result> S3FileSystemFactory::Create( + const std::string&, const std::map& options) const { + PAIMON_RETURN_NOT_OK(ValidateS3Options(options)); + return std::make_unique(options); +} + +REGISTER_PAIMON_FACTORY(S3FileSystemFactory); + +} // namespace paimon::s3 diff --git a/src/paimon/fs/s3/s3_file_system_factory.h b/src/paimon/fs/s3/s3_file_system_factory.h new file mode 100644 index 000000000..d6df28974 --- /dev/null +++ b/src/paimon/fs/s3/s3_file_system_factory.h @@ -0,0 +1,33 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "paimon/fs/file_system_factory.h" + +namespace paimon::s3 { + +class S3FileSystemFactory : 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::s3 diff --git a/src/paimon/fs/s3/s3_file_system_test.cpp b/src/paimon/fs/s3/s3_file_system_test.cpp new file mode 100644 index 000000000..a5a944099 --- /dev/null +++ b/src/paimon/fs/s3/s3_file_system_test.cpp @@ -0,0 +1,326 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/fs/s3/s3_file_system.h" + +#include + +#include +#include +#include +#include +#include +#include + +#include "paimon/common/utils/string_utils.h" +#include "paimon/fs/s3/s3_file_system_factory.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::s3 { +namespace { + +class MockHttpClient : public object_store::HttpClient { + public: + Result Execute( + const object_store::HttpRequest& request, + const object_store::HttpBodyConsumer& consumer) const override { + request_ = request; + object_store::HttpResponse response; + response.status_code = status_code_; + response.headers = response_headers_; + if (!body_.empty()) { + PAIMON_RETURN_NOT_OK(consumer(body_.data(), body_.size())); + response.body_size = body_.size(); + } + return response; + } + + mutable object_store::HttpRequest request_; + int status_code_ = 200; + object_store::HttpHeaders response_headers_; + std::string body_; +}; + +class ScopedEnvironmentVariable { + public: + ScopedEnvironmentVariable(const char* name, std::optional value) : name_(name) { + const char* previous = std::getenv(name); + if (previous != nullptr) { + previous_ = previous; + } + if (value) { + setenv(name, value->c_str(), 1); + } else { + unsetenv(name); + } + } + + ~ScopedEnvironmentVariable() { + if (previous_) { + setenv(name_.c_str(), previous_->c_str(), 1); + } else { + unsetenv(name_.c_str()); + } + } + + private: + std::string name_; + std::optional previous_; +}; + +std::map StaticOptions() { + return {{kS3AccessKeyIdOption, "access"}, + {kS3SecretAccessKeyOption, "secret"}, + {kS3SessionTokenOption, "token"}, + {kS3RegionOption, "ap-northeast-2"}}; +} + +const std::string* FindHeader(const object_store::HttpHeaders& headers, const std::string& name) { + std::string normalized_name = StringUtils::ToLowerCase(name); + for (const auto& [header_name, value] : headers) { + if (StringUtils::ToLowerCase(header_name) == normalized_name) { + return &value; + } + } + return nullptr; +} + +TEST(S3ObjectStoreClientTest, TestHeadAndSigning) { + auto http = std::make_shared(); + http->response_headers_["content-length"] = "12"; + auto client = MakeS3ObjectStoreClient(StaticOptions(), http); + ASSERT_OK_AND_ASSIGN(auto metadata, client->HeadObject({"bucket", "a b/file"})); + ASSERT_EQ(metadata.size, 12); + ASSERT_EQ(http->request_.url, "https://bucket.s3.ap-northeast-2.amazonaws.com/a%20b/file"); + const std::string* authorization = FindHeader(http->request_.headers, "authorization"); + ASSERT_NE(authorization, nullptr); + ASSERT_NE(authorization->find("/ap-northeast-2/s3/aws4_request"), std::string::npos); + ASSERT_NE(authorization->find( + "SignedHeaders=host;x-amz-content-sha256;x-amz-date;x-amz-security-token"), + std::string::npos); + ASSERT_EQ(http->request_.headers["host"], "bucket.s3.ap-northeast-2.amazonaws.com"); + const std::string* session_token = FindHeader(http->request_.headers, "x-amz-security-token"); + ASSERT_NE(session_token, nullptr); + ASSERT_EQ(*session_token, "token"); +} + +TEST(S3ObjectStoreClientTest, TestDottedBucketUsesPathStyleForDefaultEndpoint) { + auto http = std::make_shared(); + http->response_headers_["content-length"] = "12"; + auto client = MakeS3ObjectStoreClient(StaticOptions(), http); + ASSERT_OK(client->HeadObject({"paimon.prod.data", "file"})); + ASSERT_EQ(http->request_.url, "https://s3.ap-northeast-2.amazonaws.com/paimon.prod.data/file"); + ASSERT_EQ(http->request_.headers["host"], "s3.ap-northeast-2.amazonaws.com"); + + auto virtual_hosted_options = StaticOptions(); + virtual_hosted_options[kS3PathStyleAccessOption] = "false"; + client = MakeS3ObjectStoreClient(virtual_hosted_options, http); + ASSERT_OK(client->HeadObject({"paimon.prod.data", "file"})); + ASSERT_EQ(http->request_.url, "https://paimon.prod.data.s3.ap-northeast-2.amazonaws.com/file"); +} + +TEST(S3ObjectStoreClientTest, TestInvalidContentLength) { + for (const std::string content_length : + {"", "invalid", "-1", "12abc", "999999999999999999999999"}) { + auto http = std::make_shared(); + http->response_headers_["content-length"] = content_length; + auto client = MakeS3ObjectStoreClient(StaticOptions(), http); + ASSERT_TRUE(client->HeadObject({"bucket", "file"}).status().IsIOError()); + } +} + +TEST(S3ObjectStoreClientTest, TestInvalidModificationTime) { + auto http = std::make_shared(); + http->response_headers_["content-length"] = "12"; + http->response_headers_["last-modified"] = "invalid"; + auto client = MakeS3ObjectStoreClient(StaticOptions(), http); + ASSERT_OK_AND_ASSIGN(auto metadata, client->HeadObject({"bucket", "file"})); + ASSERT_EQ(metadata.modification_time, 0); + + http->body_ = + "falsefile" + "invalid12"; + ASSERT_OK_AND_ASSIGN(auto result, client->ListObjects({"bucket", ""}, "", 0)); + ASSERT_EQ(result.objects[0].modification_time, 0); +} + +TEST(S3ObjectStoreClientTest, TestRegionFromEnvironment) { + ScopedEnvironmentVariable region("AWS_REGION", "eu-west-1"); + auto options = StaticOptions(); + options.erase(kS3RegionOption); + auto http = std::make_shared(); + http->response_headers_["content-length"] = "12"; + auto client = MakeS3ObjectStoreClient(options, http); + ASSERT_OK(client->HeadObject({"bucket", "file"})); + ASSERT_EQ(http->request_.url, "https://bucket.s3.eu-west-1.amazonaws.com/file"); +} + +TEST(S3ObjectStoreClientTest, TestRegionFromProfile) { + auto test_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_dir); + std::filesystem::path config_path = std::filesystem::path(test_dir->Str()) / "region-config"; + { + std::ofstream config(config_path); + config << "[profile test-profile]\nregion = ap-south-1\n"; + } + ScopedEnvironmentVariable region("AWS_REGION", std::nullopt); + ScopedEnvironmentVariable default_region("AWS_DEFAULT_REGION", std::nullopt); + ScopedEnvironmentVariable config_file("AWS_CONFIG_FILE", config_path.string()); + auto options = StaticOptions(); + options.erase(kS3RegionOption); + options[kS3ProfileOption] = "test-profile"; + auto http = std::make_shared(); + http->response_headers_["content-length"] = "12"; + auto client = MakeS3ObjectStoreClient(options, http); + ASSERT_OK(client->HeadObject({"bucket", "file"})); + ASSERT_EQ(http->request_.url, "https://bucket.s3.ap-south-1.amazonaws.com/file"); +} + +TEST(S3ObjectStoreClientTest, TestCredentialsFromEnvironmentProfile) { + auto test_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_dir); + std::filesystem::path credentials_path = + std::filesystem::path(test_dir->Str()) / "credentials-config"; + { + std::ofstream credentials(credentials_path); + credentials << "[environment-profile]\n" + "aws_access_key_id = profile-access\n" + "aws_secret_access_key = profile-secret\n" + "aws_session_token = profile-token\n" + "[default]\n" + "aws_access_key_id = default-access\n" + "aws_secret_access_key = default-secret\n" + "aws_session_token = default-token\n"; + } + ScopedEnvironmentVariable credentials_file("AWS_SHARED_CREDENTIALS_FILE", + credentials_path.string()); + ScopedEnvironmentVariable access_key("AWS_ACCESS_KEY_ID", std::nullopt); + ScopedEnvironmentVariable secret_key("AWS_SECRET_ACCESS_KEY", std::nullopt); + ScopedEnvironmentVariable session_token("AWS_SESSION_TOKEN", std::nullopt); + { + ScopedEnvironmentVariable profile("AWS_PROFILE", "environment-profile"); + auto http = std::make_shared(); + http->response_headers_["content-length"] = "12"; + auto client = MakeS3ObjectStoreClient({{kS3RegionOption, "ap-northeast-2"}}, http); + ASSERT_OK(client->HeadObject({"bucket", "file"})); + const std::string* authorization = FindHeader(http->request_.headers, "authorization"); + ASSERT_NE(authorization, nullptr); + ASSERT_NE(authorization->find("Credential=profile-access/"), std::string::npos); + const std::string* token = FindHeader(http->request_.headers, "x-amz-security-token"); + ASSERT_NE(token, nullptr); + ASSERT_EQ(*token, "profile-token"); + } + { + ScopedEnvironmentVariable profile("AWS_PROFILE", std::nullopt); + auto http = std::make_shared(); + http->response_headers_["content-length"] = "12"; + auto client = MakeS3ObjectStoreClient({{kS3RegionOption, "ap-northeast-2"}}, http); + ASSERT_OK(client->HeadObject({"bucket", "file"})); + const std::string* authorization = FindHeader(http->request_.headers, "authorization"); + ASSERT_NE(authorization, nullptr); + ASSERT_NE(authorization->find("Credential=default-access/"), std::string::npos); + const std::string* token = FindHeader(http->request_.headers, "x-amz-security-token"); + ASSERT_NE(token, nullptr); + ASSERT_EQ(*token, "default-token"); + } +} + +TEST(S3ObjectStoreClientTest, TestRangeAndListObjects) { + auto http = std::make_shared(); + auto client = MakeS3ObjectStoreClient(StaticOptions(), http); + http->body_ = "data"; + char buffer[4]; + ASSERT_OK_AND_ASSIGN(auto size, client->GetObjectRange({"bucket", "key"}, 2, 4, buffer)); + ASSERT_EQ(size, 4); + ASSERT_EQ(std::string(buffer, sizeof(buffer)), "data"); + ASSERT_EQ(http->request_.headers["range"], "bytes=2-5"); + + http->body_ = + "true" + "dir/a&b2026-01-01T00:00:00Z" + "7dir/a&lt;b8" + "dir/sub/" + "next token"; + ASSERT_OK_AND_ASSIGN(auto result, client->ListObjects({"bucket", "dir/"}, "old token", 10)); + 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[1].key, "dir/a<b"); + 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); + ASSERT_NE(http->request_.url.find("continuation-token=old%20token"), std::string::npos); +} + +TEST(S3ObjectStoreClientTest, TestUrlEncodedListObjects) { + auto http = std::make_shared(); + auto client = MakeS3ObjectStoreClient(StaticOptions(), http); + http->body_ = + "true" + "dir/a%26b%0D7" + "dir/sub%25/" + "next%2Ftoken"; + ASSERT_OK_AND_ASSIGN(auto result, client->ListObjects({"bucket", "dir/"}, "", 0)); + ASSERT_EQ(result.objects[0].key, "dir/a&b\r"); + ASSERT_EQ(result.common_prefixes[0], "dir/sub%/"); + ASSERT_EQ(result.continuation_token, "next/token"); + + http->body_ = + "false" + "dir/invalid%27"; + ASSERT_TRUE(client->ListObjects({"bucket", "dir/"}, "", 0).status().IsIOError()); +} + +TEST(S3ObjectStoreClientTest, TestInvalidListObjectsResponse) { + auto http = std::make_shared(); + auto client = MakeS3ObjectStoreClient(StaticOptions(), http); + for (const std::string size : {"", "abc", "-1", "12abc", "9223372036854775808"}) { + http->body_ = + "falsekey" + "" + + size + ""; + ASSERT_TRUE(client->ListObjects({"bucket", ""}, "", 0).status().IsIOError()); + } + http->body_ = + "falsekey" + "1"; + ASSERT_TRUE(client->ListObjects({"bucket", ""}, "", 0).status().IsIOError()); + http->body_ = + "key1"; + ASSERT_TRUE(client->ListObjects({"bucket", ""}, "", 0).status().IsIOError()); +} + +TEST(S3FileSystemFactoryTest, TestOptionValidation) { + S3FileSystemFactory factory; + ASSERT_TRUE(factory + .Create("s3://bucket", + {{kS3AccessKeyIdOption, ""}, {kS3SecretAccessKeyOption, "secret"}}) + .status() + .IsInvalid()); + ASSERT_TRUE( + factory.Create("s3://bucket", {{kS3PathStyleAccessOption, "treu"}}).status().IsInvalid()); + ASSERT_TRUE( + factory.Create("s3://bucket", {{kS3SessionTokenOption, "token"}}).status().IsInvalid()); + + auto http = std::make_shared(); + auto client = MakeS3ObjectStoreClient({{kS3AccessKeyIdOption, "access"}}, http); + ASSERT_TRUE(client->HeadObject({"bucket", "file"}).status().IsInvalid()); + client = MakeS3ObjectStoreClient({{kS3PathStyleAccessOption, "treu"}}, http); + ASSERT_TRUE(client->HeadObject({"bucket", "file"}).status().IsInvalid()); +} + +} // namespace +} // namespace paimon::s3 diff --git a/third_party/versions.txt b/third_party/versions.txt index 20d9e9d27..5a557ccf3 100644 --- a/third_party/versions.txt +++ b/third_party/versions.txt @@ -72,6 +72,32 @@ PAIMON_AVRO_BUILD_VERSION=c499eefb48aa2db906c7bca14a047223806f36db PAIMON_AVRO_BUILD_SHA256_CHECKSUM=9771f1dcfe3c01aff7ff670e873e66d3406362f71941821d482de65f3d32d780 PAIMON_AVRO_PKG_NAME=avro-${PAIMON_AVRO_BUILD_VERSION}.tar.gz +PAIMON_AWS_C_AUTH_BUILD_VERSION=v0.10.3 +PAIMON_AWS_C_AUTH_BUILD_SHA256_CHECKSUM=20fc5e75529fadd81fd38b25f9d83798b53ab235ebbac92cdfbb716cfcc7593d +PAIMON_AWS_C_AUTH_PKG_NAME=aws-c-auth-${PAIMON_AWS_C_AUTH_BUILD_VERSION}.tar.gz +PAIMON_AWS_C_CAL_BUILD_VERSION=v0.9.14 +PAIMON_AWS_C_CAL_BUILD_SHA256_CHECKSUM=0e96e0067fa921768e07b5b4ebad82011ccf474903e9286419ef428d68f317ea +PAIMON_AWS_C_CAL_PKG_NAME=aws-c-cal-${PAIMON_AWS_C_CAL_BUILD_VERSION}.tar.gz +PAIMON_AWS_C_COMMON_BUILD_VERSION=v0.14.0 +PAIMON_AWS_C_COMMON_BUILD_SHA256_CHECKSUM=3684076ec5da899074336722ba58a01f7166a1a2e5ad72f846f6fd468ecdf2ec +PAIMON_AWS_C_COMMON_PKG_NAME=aws-c-common-${PAIMON_AWS_C_COMMON_BUILD_VERSION}.tar.gz +PAIMON_AWS_C_COMPRESSION_BUILD_VERSION=v0.3.2 +PAIMON_AWS_C_COMPRESSION_BUILD_SHA256_CHECKSUM=f93f5a5d8b3fee3a6d97b14ba279efacd4d4016ef9cc7dc4be7d43519ecfbe93 +PAIMON_AWS_C_COMPRESSION_PKG_NAME=aws-c-compression-${PAIMON_AWS_C_COMPRESSION_BUILD_VERSION}.tar.gz +PAIMON_AWS_C_HTTP_BUILD_VERSION=v0.11.0 +PAIMON_AWS_C_HTTP_BUILD_SHA256_CHECKSUM=4ccbdd33c798b590288330dec9e93abe2ff6cfb198b7a4db036c9d362f2e6506 +PAIMON_AWS_C_HTTP_PKG_NAME=aws-c-http-${PAIMON_AWS_C_HTTP_BUILD_VERSION}.tar.gz +PAIMON_AWS_C_IO_BUILD_VERSION=v0.27.0 +PAIMON_AWS_C_IO_BUILD_SHA256_CHECKSUM=e89a1f784e7c97e4197031ffdcf30f67d66d7c14f8a391edf5764f17dae982ee +PAIMON_AWS_C_IO_PKG_NAME=aws-c-io-${PAIMON_AWS_C_IO_BUILD_VERSION}.tar.gz +PAIMON_AWS_C_SDKUTILS_BUILD_VERSION=v0.2.5 +PAIMON_AWS_C_SDKUTILS_BUILD_SHA256_CHECKSUM=13a03ea87aa67c7db414bf245fbcc623555c783a34d8ba1d7d701fd42717c366 +PAIMON_AWS_C_SDKUTILS_PKG_NAME=aws-c-sdkutils-${PAIMON_AWS_C_SDKUTILS_BUILD_VERSION}.tar.gz +PAIMON_AWS_S2N_BUILD_VERSION=v1.7.4 +PAIMON_AWS_S2N_BUILD_REVISION=eaf2c08a78f7fca6279e3704d3fa1c4e6c9a6abc +PAIMON_AWS_S2N_BUILD_SHA256_CHECKSUM=59cacc899538668345685e107cb80cb96144a77c482fbd51cc0620e344773100 +PAIMON_AWS_S2N_PKG_NAME=s2n-${PAIMON_AWS_S2N_BUILD_VERSION}.zip + PAIMON_FMT_BUILD_VERSION=11.2.0 PAIMON_FMT_BUILD_SHA256_CHECKSUM=bc23066d87ab3168f27cef3e97d545fa63314f5c79df5ea444d41d56f962c6af PAIMON_FMT_PKG_NAME=fmt-${PAIMON_FMT_BUILD_VERSION}.tar.gz