feat(fs): enhance S3 filesystem reliability and compatibility

CMakeLists.txt

@@ -487,9 +487,6 @@ install(FILES "${CMAKE_CURRENT_BINARY_DIR}/PaimonConfig.cmake"
 
 config_summary_message()
 
-if(PAIMON_ENABLE_S3)
-    find_package(CURL REQUIRED)
-endif()
 add_subdirectory(src/paimon)
 add_subdirectory(src/paimon/fs/local)
 add_subdirectory(src/paimon/fs/jindo)

cmake_modules/ThirdpartyToolchain.cmake

@@ -1890,6 +1890,20 @@ endif()
 if(PAIMON_ENABLE_S3)
     include(BuildAwsAuth)
     build_aws_auth()
+
+    find_package(CURL 7.62 REQUIRED)
+    set(PAIMON_CURL_IS_STATIC OFF)
+    foreach(PAIMON_CURL_LIBRARY IN LISTS CURL_LIBRARIES)
+        if(PAIMON_CURL_LIBRARY MATCHES "\\${CMAKE_STATIC_LIBRARY_SUFFIX}$")
+            set(PAIMON_CURL_IS_STATIC ON)
+            break()
+        endif()
+    endforeach()
+
+    set(PAIMON_HTTP_LINK_LIBS CURL::libcurl)
+    if(PAIMON_CURL_IS_STATIC)
+        list(APPEND PAIMON_HTTP_LINK_LIBS OpenSSL::SSL OpenSSL::Crypto)
+    endif()
 endif()
 if(PAIMON_ENABLE_LUMINA)
     build_lumina()

src/paimon/CMakeLists.txt

@@ -181,7 +181,6 @@ set(PAIMON_COMMON_SRCS
 if(PAIMON_ENABLE_S3)
     list(APPEND PAIMON_COMMON_SRCS common/fs/http_client.cpp
          common/fs/object_store_file_system.cpp)
-    set(PAIMON_OBJECT_STORE_LINK_LIBS CURL::libcurl)
 endif()
 
 set(PAIMON_CORE_SRCS
@@ -416,7 +415,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
-               ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_HTTP_LINK_LIBS}
                STATIC_LINK_LIBS
                arrow
                tbb
@@ -426,7 +425,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
-               ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_HTTP_LINK_LIBS}
                SHARED_LINK_FLAGS
                ${PAIMON_VERSION_SCRIPT_FLAGS})
 
@@ -852,6 +851,14 @@ if(PAIMON_BUILD_TESTS)
     if(PAIMON_ENABLE_S3)
         list(APPEND PAIMON_OBJECT_STORE_FS_TEST_SOURCES
              common/fs/object_store_file_system_test.cpp)
+
+        add_paimon_test(http_client_test
+                        SOURCES
+                        common/fs/http_client_test.cpp
+                        STATIC_LINK_LIBS
+                        paimon_shared
+                        fmt
+                        ${GTEST_LINK_TOOLCHAIN})
     endif()
 
     add_paimon_test(fs_test

src/paimon/common/fs/http_client.cpp

@@ -18,6 +18,7 @@
 
 #include <curl/curl.h>
 
+#include <algorithm>
 #include <chrono>
 #include <mutex>
 #include <thread>
@@ -33,6 +34,7 @@ namespace paimon {
 namespace {
 
 constexpr int32_t kMaxAttempts = 3;
+constexpr size_t kMaxErrorBodySize = 4096;
 
 class CurlGlobalGuard {
  public:
@@ -78,6 +80,8 @@ size_t WriteCallback(char* data, size_t size, size_t count, void* user_data) {
     curl_easy_getinfo(context->handle, CURLINFO_RESPONSE_CODE, &status_code);
     context->response.status_code = static_cast<int32_t>(status_code);
     if (status_code >= 300) {
+        size_t remaining = kMaxErrorBodySize - context->response.error_body.size();
+        context->response.error_body.append(data, std::min(bytes, remaining));
         context->response.body_size += static_cast<int64_t>(bytes);
         return bytes;
     }
@@ -118,7 +122,8 @@ bool IsRetryable(CURLcode code, int64_t status_code) {
 
 class CurlHttpClient::Impl {
  public:
-    Impl() : guard_(GetCurlGlobalGuard()) {}
+    explicit Impl(CurlHttpClientOptions options)
+        : guard_(GetCurlGlobalGuard()), options_(std::move(options)) {}
 
     ~Impl() {
         for (CURL* handle : handles_) {
@@ -142,13 +147,19 @@ class CurlHttpClient::Impl {
         handles_.push_back(handle);
     }
 
+    const CurlHttpClientOptions& options() const {
+        return options_;
+    }
+
  private:
     std::shared_ptr<CurlGlobalGuard> guard_;
+    CurlHttpClientOptions options_;
     mutable std::mutex mutex_;
     mutable std::vector<CURL*> handles_;
 };
 
-CurlHttpClient::CurlHttpClient() : impl_(std::make_unique<Impl>()) {}
+CurlHttpClient::CurlHttpClient(CurlHttpClientOptions options)
+    : impl_(std::make_unique<Impl>(std::move(options))) {}
 CurlHttpClient::~CurlHttpClient() = default;
 
 Result<HttpResponse> CurlHttpClient::Execute(const HttpRequest& request,
@@ -172,7 +183,18 @@ Result<HttpResponse> CurlHttpClient::Execute(const HttpRequest& request,
         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_CONNECTTIMEOUT_MS,
+            static_cast<long>(impl_->options().connect_timeout_ms));  // NOLINT(runtime/int)
+        curl_easy_setopt(
+            handle, CURLOPT_TIMEOUT_MS,
+            static_cast<long>(impl_->options().request_timeout_ms));  // NOLINT(runtime/int)
+        curl_easy_setopt(handle, CURLOPT_LOW_SPEED_LIMIT,
+                         static_cast<long>(  // NOLINT(runtime/int): required by curl.
+                             impl_->options().low_speed_limit_bytes_per_second));
+        curl_easy_setopt(handle, CURLOPT_LOW_SPEED_TIME,
+                         static_cast<long>(  // NOLINT(runtime/int): required by curl.
+                             impl_->options().low_speed_time_seconds));
         curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, WriteCallback);
         curl_easy_setopt(handle, CURLOPT_WRITEDATA, &context);
         curl_easy_setopt(handle, CURLOPT_HEADERFUNCTION, HeaderCallback);
@@ -196,8 +218,7 @@ Result<HttpResponse> CurlHttpClient::Execute(const HttpRequest& request,
             (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 context.response;
             }
             return Status::IOError(fmt::format("HTTP request to {} failed: {} (status {})",
                                                request.url, curl_easy_strerror(code),

src/paimon/common/fs/http_client.h

@@ -41,6 +41,7 @@ struct HttpResponse {
     int32_t status_code = 0;
     HttpHeaders headers;
     int64_t body_size = 0;
+    std::string error_body;
 };
 
 class PAIMON_EXPORT HttpClient {
@@ -50,9 +51,16 @@ class PAIMON_EXPORT HttpClient {
                                          const HttpBodyConsumer& consumer) const = 0;
 };
 
+struct CurlHttpClientOptions {
+    int64_t connect_timeout_ms = 30000;
+    int64_t request_timeout_ms = 300000;
+    int64_t low_speed_limit_bytes_per_second = 1;
+    int64_t low_speed_time_seconds = 30;
+};
+
 class PAIMON_EXPORT CurlHttpClient : public HttpClient {
  public:
-    CurlHttpClient();
+    explicit CurlHttpClient(CurlHttpClientOptions options = {});
     ~CurlHttpClient() override;
 
     Result<HttpResponse> Execute(const HttpRequest& request,

src/paimon/common/fs/http_client_test.cpp

@@ -0,0 +1,214 @@
+/*
+ * 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/common/fs/http_client.h"
+
+#include <arpa/inet.h>
+#include <gtest/gtest.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <chrono>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <optional>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+class ScopedEnvironmentVariable {
+ public:
+    ScopedEnvironmentVariable(const char* name, const char* value) : name_(name) {
+        const char* previous = std::getenv(name);
+        if (previous != nullptr) {
+            previous_ = previous;
+        }
+        setenv(name, value, 1);
+    }
+
+    ~ScopedEnvironmentVariable() {
+        if (previous_) {
+            setenv(name_.c_str(), previous_->c_str(), 1);
+        } else {
+            unsetenv(name_.c_str());
+        }
+    }
+
+ private:
+    std::string name_;
+    std::optional<std::string> previous_;
+};
+
+class HttpTestServer {
+ public:
+    explicit HttpTestServer(std::vector<std::string> responses,
+                            std::chrono::milliseconds response_delay = {})
+        : responses_(std::move(responses)), response_delay_(response_delay) {
+        socket_ = socket(AF_INET, SOCK_STREAM, 0);
+        if (socket_ < 0) {
+            return;
+        }
+        int32_t reuse_address = 1;
+        setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, &reuse_address, sizeof(reuse_address));
+        sockaddr_in address{};
+        address.sin_family = AF_INET;
+        address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+        address.sin_port = 0;
+        if (bind(socket_, reinterpret_cast<sockaddr*>(&address), sizeof(address)) != 0 ||
+            listen(socket_, 4) != 0) {
+            close(socket_);
+            socket_ = -1;
+            return;
+        }
+        socklen_t address_size = sizeof(address);
+        if (getsockname(socket_, reinterpret_cast<sockaddr*>(&address), &address_size) != 0) {
+            close(socket_);
+            socket_ = -1;
+            return;
+        }
+        port_ = ntohs(address.sin_port);
+        thread_ = std::thread([this] { Serve(); });
+    }
+
+    ~HttpTestServer() {
+        if (socket_ >= 0) {
+            shutdown(socket_, SHUT_RDWR);
+            close(socket_);
+        }
+        if (thread_.joinable()) {
+            thread_.join();
+        }
+    }
+
+    bool ok() const {
+        return socket_ >= 0;
+    }
+
+    std::string url() const {
+        return fmt::format("http://127.0.0.1:{}/test", port_);
+    }
+
+ private:
+    void Serve() const {
+        for (const std::string& response : responses_) {
+            int32_t client = accept(socket_, nullptr, nullptr);
+            if (client < 0) {
+                return;
+            }
+            char request[4096];
+            recv(client, request, sizeof(request), 0);
+            if (response_delay_.count() > 0) {
+                std::this_thread::sleep_for(response_delay_);
+            }
+            size_t sent = 0;
+            while (sent < response.size()) {
+                ssize_t result = send(client, response.data() + sent, response.size() - sent, 0);
+                if (result <= 0) {
+                    break;
+                }
+                sent += static_cast<size_t>(result);
+            }
+            shutdown(client, SHUT_RDWR);
+            close(client);
+        }
+    }
+
+    int32_t socket_ = -1;
+    uint16_t port_ = 0;
+    std::vector<std::string> responses_;
+    std::chrono::milliseconds response_delay_;
+    std::thread thread_;
+};
+
+TEST(CurlHttpClientTest, TestCapturesBoundedErrorBodyAndHeaders) {
+    ScopedEnvironmentVariable no_proxy("NO_PROXY", "127.0.0.1");
+    ScopedEnvironmentVariable lowercase_no_proxy("no_proxy", "127.0.0.1");
+    std::string body(5000, 'x');
+    HttpTestServer server(
+        {"HTTP/1.1 403 Forbidden\r\nContent-Length: 5000\r\n"
+         "X-Amz-Bucket-Region: eu-west-1\r\nConnection: close\r\n\r\n" +
+         body});
+    ASSERT_TRUE(server.ok());
+
+    CurlHttpClient client;
+    bool consumer_called = false;
+    ASSERT_OK_AND_ASSIGN(HttpResponse response,
+                         client.Execute({HttpMethod::GET, server.url(), {}},
+                                        [&consumer_called](const char*, int64_t) {
+                                            consumer_called = true;
+                                            return Status::OK();
+                                        }));
+    ASSERT_EQ(response.status_code, 403);
+    ASSERT_EQ(response.body_size, 5000);
+    ASSERT_EQ(response.error_body.size(), 4096);
+    ASSERT_EQ(response.headers["x-amz-bucket-region"], "eu-west-1");
+    ASSERT_FALSE(consumer_called);
+}
+
+TEST(CurlHttpClientTest, TestConfiguredRequestTimeout) {
+    ScopedEnvironmentVariable no_proxy("NO_PROXY", "127.0.0.1");
+    ScopedEnvironmentVariable lowercase_no_proxy("no_proxy", "127.0.0.1");
+    HttpTestServer server({"", "", ""}, std::chrono::milliseconds(100));
+    ASSERT_TRUE(server.ok());
+    CurlHttpClientOptions options;
+    options.connect_timeout_ms = 100;
+    options.request_timeout_ms = 50;
+    options.low_speed_limit_bytes_per_second = 0;
+    options.low_speed_time_seconds = 0;
+    CurlHttpClient client(options);
+
+    auto start = std::chrono::steady_clock::now();
+    Result<HttpResponse> response = client.Execute(
+        {HttpMethod::GET, server.url(), {}}, [](const char*, int64_t) { return Status::OK(); });
+    auto elapsed = std::chrono::steady_clock::now() - start;
+    ASSERT_TRUE(response.status().IsIOError());
+    ASSERT_LT(elapsed, std::chrono::seconds(3));
+}
+
+TEST(CurlHttpClientTest, TestRetriesServerErrors) {
+    ScopedEnvironmentVariable no_proxy("NO_PROXY", "127.0.0.1");
+    ScopedEnvironmentVariable lowercase_no_proxy("no_proxy", "127.0.0.1");
+    std::string server_error =
+        "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 5\r\n"
+        "Connection: close\r\n\r\nerror";
+    HttpTestServer server(
+        {server_error, server_error,
+         "HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\ndata"});
+    ASSERT_TRUE(server.ok());
+
+    CurlHttpClient client;
+    std::string body;
+    ASSERT_OK_AND_ASSIGN(HttpResponse response,
+                         client.Execute({HttpMethod::GET, server.url(), {}},
+                                        [&body](const char* data, int64_t size) {
+                                            body.append(data, static_cast<size_t>(size));
+                                            return Status::OK();
+                                        }));
+    ASSERT_EQ(response.status_code, 200);
+    ASSERT_EQ(body, "data");
+}
+
+}  // namespace
+}  // namespace paimon::test