feat(rest): support rest catalog with database/table operations and snapshot listing

Add a REST catalog implementation selected via the "metastore=rest"
option:

- RestHttpClient: blocking libcurl client with exponential-backoff
  retries (429/503 retried for all methods, transport errors only
  for idempotent ones, honoring a Retry-After header in both the
  delta-seconds and the HTTP-date form when it yields a positive
  delay, with a locale-independent date parse; a server closing the
  connection without responding is not retried) and per-request
  debug logging.
- RestApi: HTTP + JSON layer with "/v1/config" option merging
  (overrides > client options > defaults), "header." options as
  request headers, paged listing and error-to-Status mapping that
  prefers the code of the parsed error body over the http status,
  redacts sensitive server messages, never echoes response bodies
  into error messages (they may carry credentials), carries the
  request id (x-request-id, falling back to any request-id header)
  and attaches a RestErrorDetail with the mapped code so callers can
  distinguish e.g. authentication failures programmatically.
- RestCatalog: database/table operations and snapshot listing,
  "table-default." option defaults, system/branch table checks and
  schema conversion with the computed highestFieldId, rejecting
  duplicated field ids at any nesting level. Branch identifiers are
  passed through to the server so it resolves the branch and returns
  the branch's own schema (the default branch "main" resolves to the
  bare table), snapshots of a branch are listed under the branch
  object name, and the "sys" database serves the local global system
  tables like FileSystemCatalog.
- Bear token authentication provider.
- CatalogOptions: catalog-level option keys (metastore, uri, token,
  token.provider) and the "table-default." option prefix in a
  dedicated public header, keeping them separate from the table-level
  CoreOptions.
- CatalogUtils: system database and system/branch table checks
  shared by FileSystemCatalog and RestCatalog; FileSystemCatalog's
  create/drop/rename checks now use them instead of inline messages.
- SpecialFieldIds::SYSTEM_FIELD_ID_START names the reserved system
  field id bound, which is excluded from the highest field id.

The pieces overlapping with the object store file systems are unified
under common/utils: UrlUtils carries both URL encoding flavors
(form-urlencoded for the REST api, RFC 3986 for S3), the generic HTTP
client moved there from common/fs, and the rest client reuses its
curl global-init guard, ScopeGuard cleanup and header-line parsing.
libcurl detection in CMake is a single find_package guarded by
PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST.

libcurl is used from the system rather than bundled, so the new
PAIMON_ENABLE_REST option defaults to OFF like the other optional
components with external requirements and is documented alongside
them.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: SteNicholas

CMakeLists.txt

@@ -66,9 +66,17 @@ option(PAIMON_ENABLE_LUMINA "Whether to enable lumina vector index" OFF)
 option(PAIMON_ENABLE_LUCENE "Whether to enable lucene index" OFF)
 option(PAIMON_ENABLE_TANTIVY
        "Whether to enable tantivy-fulltext global index (Rust FFI, experimental)" OFF)
+option(PAIMON_ENABLE_REST "Whether to enable the rest catalog (requires libcurl)" OFF)
 if(PAIMON_ENABLE_ORC)
     add_definitions(-DPAIMON_ENABLE_ORC)
 endif()
+if(PAIMON_ENABLE_REST)
+    add_definitions(-DPAIMON_ENABLE_REST)
+endif()
+# libcurl backs the HTTP client shared by the S3 file system and the rest catalog.
+if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST)
+    find_package(CURL REQUIRED)
+endif()
 if(PAIMON_ENABLE_AVRO)
     add_definitions(-DPAIMON_ENABLE_AVRO)
 endif()
@@ -487,9 +495,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)

ci/scripts/build_paimon.sh

@@ -130,6 +130,7 @@ CMAKE_ARGS=(
     "-DPAIMON_ENABLE_LUMINA=${ENABLE_LUMINA}"
     "-DPAIMON_ENABLE_LUCENE=ON"
     "-DPAIMON_ENABLE_TANTIVY=${ENABLE_TANTIVY}"
+    "-DPAIMON_ENABLE_REST=ON"
     "-DPAIMON_LINT_GIT_TARGET_COMMIT=${lint_git_target_commit}"
 )
 

docs/source/building.rst

@@ -145,6 +145,7 @@ boolean flags to ``cmake``.
 * ``-DPAIMON_ENABLE_AVRO=ON``: Apache Avro libraries and Paimon integration
 * ``-DPAIMON_ENABLE_JINDO=ON``: Support for Alibaba Jindo filesystems
 * ``-DPAIMON_ENABLE_LUMINA=ON``: Support for Lumina vector index, lumina is only supported on gcc9 or higher.
+* ``-DPAIMON_ENABLE_REST=ON``: Support for the REST catalog (``metastore=rest``), requires the libcurl development package.
 
 Third-party dependency source
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

include/paimon/catalog_options.h

@@ -0,0 +1,46 @@
+/*
+ * 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/visibility.h"
+
+namespace paimon {
+
+/// Catalog-level configuration option keys; table-level keys live in `Options`.
+struct PAIMON_EXPORT CatalogOptions {
+    /// "metastore" - Metastore of the paimon catalog.
+    /// Supported values are "filesystem" (default) and "rest".
+    static const char METASTORE[];
+
+    /// "uri" - Server url of the REST catalog. Only used when METASTORE is "rest".
+    static const char URI[];
+
+    /// "token" - Token of the "bear" token provider of the REST catalog.
+    static const char TOKEN[];
+
+    /// "token.provider" - Authentication provider of the REST catalog.
+    /// Currently only "bear" is supported ("bear" is the protocol's historical
+    /// spelling of "bearer", do not "fix" it).
+    static const char TOKEN_PROVIDER[];
+
+    /// "table-default." - Prefix of the catalog options that provide table option
+    /// defaults: "table-default.<key>=<value>" applies "<key>=<value>" to a created
+    /// table when the caller left "<key>" unset.
+    static const char TABLE_DEFAULT_OPTION_PREFIX[];
+};
+
+}  // namespace paimon

include/paimon/utils/special_field_ids.h

@@ -39,6 +39,10 @@ class SpecialFieldIds {
 
     /// Special field ID reserved for index score. Value: CPP_FIELD_ID_END - 1
     static const int32_t INDEX_SCORE = CPP_FIELD_ID_END - 1;
+
+    /// The lowest field ID reserved for system fields; IDs at or above this bound are
+    /// excluded when computing the highest field ID of a schema. Value: INT32_MAX / 2
+    static const int32_t SYSTEM_FIELD_ID_START = std::numeric_limits<int32_t>::max() / 2;
 };
 
 }  // namespace paimon

src/paimon/CMakeLists.txt

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 set(PAIMON_COMMON_SRCS
+    common/catalog_options.cpp
     common/compression/block_compression_factory.cpp
     common/compression/block_compressor.cpp
     common/compression/block_decompressor.cpp
@@ -176,12 +177,18 @@ set(PAIMON_COMMON_SRCS
     common/utils/roaring_bitmap64.cpp
     common/utils/row_range_index.cpp
     common/utils/status.cpp
-    common/utils/string_utils.cpp)
+    common/utils/string_utils.cpp
+    common/utils/url_utils.cpp)
 
+# The shared HTTP client is used by both the object store file systems and the
+# rest catalog.
+set(PAIMON_CURL_LINK_LIBS)
+if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST)
+    list(APPEND PAIMON_COMMON_SRCS common/utils/http_client.cpp)
+    set(PAIMON_CURL_LINK_LIBS CURL::libcurl)
+endif()
 if(PAIMON_ENABLE_S3)
-    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)
+    list(APPEND PAIMON_COMMON_SRCS common/fs/object_store_file_system.cpp)
 endif()
 
 set(PAIMON_CORE_SRCS
@@ -224,6 +231,7 @@ set(PAIMON_CORE_SRCS
     core/casting/timestamp_to_timestamp_cast_executor.cpp
     core/casting/casting_utils.cpp
     core/catalog/catalog.cpp
+    core/catalog/catalog_utils.cpp
     core/catalog/file_system_catalog.cpp
     core/catalog/identifier.cpp
     core/core_options.cpp
@@ -403,6 +411,18 @@ set(PAIMON_CORE_SRCS
     core/utils/special_field_ids.cpp
     core/utils/tag_manager.cpp)
 
+if(PAIMON_ENABLE_REST)
+    list(APPEND
+         PAIMON_CORE_SRCS
+         rest/rest_http_client.cpp
+         rest/resource_paths.cpp
+         rest/rest_api.cpp
+         rest/rest_auth.cpp
+         rest/rest_catalog.cpp
+         rest/rest_messages.cpp
+         rest/rest_util.cpp)
+endif()
+
 add_paimon_lib(paimon
                SOURCES
                ${PAIMON_COMMON_SRCS}
@@ -416,7 +436,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
-               ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_CURL_LINK_LIBS}
                STATIC_LINK_LIBS
                arrow
                tbb
@@ -426,14 +446,21 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
-               ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_CURL_LINK_LIBS}
                SHARED_LINK_FLAGS
                ${PAIMON_VERSION_SCRIPT_FLAGS})
 
 add_subdirectory(common/file_index)
 add_subdirectory(common/global_index)
 
 if(PAIMON_BUILD_TESTS)
+    # The shared HTTP client is compiled only for the components that need libcurl,
+    # so its test follows the same gate.
+    set(PAIMON_COMMON_HTTP_CLIENT_TEST_SRCS)
+    if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST)
+        set(PAIMON_COMMON_HTTP_CLIENT_TEST_SRCS common/utils/http_client_test.cpp)
+    endif()
+
     add_paimon_test(memory_test
                     SOURCES
                     common/memory/memory_pool_test.cpp
@@ -594,6 +621,8 @@ if(PAIMON_BUILD_TESTS)
                     common/utils/status_test.cpp
                     common/utils/stream_utils_test.cpp
                     common/utils/string_utils_test.cpp
+                    common/utils/url_utils_test.cpp
+                    ${PAIMON_COMMON_HTTP_CLIENT_TEST_SRCS}
                     common/utils/range_test.cpp
                     common/utils/uuid_test.cpp
                     common/utils/decimal_utils_test.cpp
@@ -870,4 +899,18 @@ if(PAIMON_BUILD_TESTS)
                     EXTRA_INCLUDES
                     ${JINDOSDK_INCLUDE_DIR})
 
+    if(PAIMON_ENABLE_REST)
+        add_paimon_test(rest_test
+                        SOURCES
+                        rest/rest_http_client_test.cpp
+                        rest/mock_rest_server.cpp
+                        rest/rest_catalog_test.cpp
+                        rest/rest_messages_test.cpp
+                        STATIC_LINK_LIBS
+                        paimon_shared
+                        test_utils_static
+                        ${TEST_STATIC_LINK_LIBS}
+                        ${GTEST_LINK_TOOLCHAIN})
+    endif()
+
 endif()

src/paimon/common/catalog_options.cpp

@@ -0,0 +1,27 @@
+/*
+ * 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/catalog_options.h"
+
+namespace paimon {
+
+const char CatalogOptions::METASTORE[] = "metastore";
+const char CatalogOptions::URI[] = "uri";
+const char CatalogOptions::TOKEN[] = "token";
+const char CatalogOptions::TOKEN_PROVIDER[] = "token.provider";
+const char CatalogOptions::TABLE_DEFAULT_OPTION_PREFIX[] = "table-default.";
+
+}  // namespace paimon

src/paimon/common/fs/http_client.cpp src/paimon/common/utils/http_client.cppsrc/paimon/common/fs/http_client.cpp renamed to src/paimon/common/utils/http_client.cpp

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-#include "paimon/common/fs/http_client.h"
+#include "paimon/common/utils/http_client.h"
 
 #include <curl/curl.h>
 
@@ -44,11 +44,6 @@ class CurlGlobalGuard {
     }
 };
 
-std::shared_ptr<CurlGlobalGuard> GetCurlGlobalGuard() {
-    static auto guard = std::make_shared<CurlGlobalGuard>();
-    return guard;
-}
-
 void TrimHttpWhitespace(std::string* value) {
     constexpr char kHttpWhitespace[] = " \t\r\n";
     size_t begin = value->find_first_not_of(kHttpWhitespace);
@@ -92,16 +87,7 @@ size_t WriteCallback(char* data, size_t size, size_t count, void* user_data) {
 size_t HeaderCallback(char* data, size_t size, size_t count, void* user_data) {
     auto* context = static_cast<TransferContext*>(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 = line.substr(0, colon);
-        TrimHttpWhitespace(&name);
-        name = StringUtils::ToLowerCase(name);
-        std::string value = line.substr(colon + 1);
-        TrimHttpWhitespace(&value);
-        context->response.headers[name] = std::move(value);
-    }
+    ParseHttpHeaderLine(data, bytes, &context->response.headers);
     return bytes;
 }
 
@@ -116,9 +102,31 @@ bool IsRetryable(CURLcode code, int64_t status_code) {
 
 }  // namespace
 
+std::shared_ptr<void> EnsureCurlGlobalInit() {
+    static auto guard = std::make_shared<CurlGlobalGuard>();
+    return guard;
+}
+
+void ParseHttpHeaderLine(const char* data, size_t size, HttpHeaders* headers) {
+    std::string line(data, size);
+    size_t colon = line.find(':');
+    if (colon == std::string::npos) {
+        return;
+    }
+    std::string name = line.substr(0, colon);
+    TrimHttpWhitespace(&name);
+    if (name.empty()) {
+        return;
+    }
+    name = StringUtils::ToLowerCase(name);
+    std::string value = line.substr(colon + 1);
+    TrimHttpWhitespace(&value);
+    (*headers)[name] = std::move(value);
+}
+
 class CurlHttpClient::Impl {
  public:
-    Impl() : guard_(GetCurlGlobalGuard()) {}
+    Impl() : guard_(EnsureCurlGlobalInit()) {}
 
     ~Impl() {
         for (CURL* handle : handles_) {
@@ -143,7 +151,7 @@ class CurlHttpClient::Impl {
     }
 
  private:
-    std::shared_ptr<CurlGlobalGuard> guard_;
+    std::shared_ptr<void> guard_;
     mutable std::mutex mutex_;
     mutable std::vector<CURL*> handles_;
 };

src/paimon/common/fs/http_client.h src/paimon/common/utils/http_client.hsrc/paimon/common/fs/http_client.h renamed to src/paimon/common/utils/http_client.h

@@ -31,6 +31,14 @@ enum class HttpMethod { HEAD, GET };
 using HttpHeaders = std::map<std::string, std::string>;
 using HttpBodyConsumer = std::function<Status(const char*, int64_t)>;
 
+/// Ensures libcurl's global state is initialized; the returned guard keeps it alive.
+PAIMON_EXPORT std::shared_ptr<void> EnsureCurlGlobalInit();
+
+/// Parses one raw HTTP header line into `headers`, lower-casing the name and trimming
+/// HTTP whitespace around the name and value; lines without a ':' or with an empty
+/// name are ignored.
+PAIMON_EXPORT void ParseHttpHeaderLine(const char* data, size_t size, HttpHeaders* headers);
+
 struct HttpRequest {
     HttpMethod method = HttpMethod::GET;
     std::string url;

src/paimon/common/utils/http_client_test.cpp

@@ -0,0 +1,47 @@
+/*
+ * 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/utils/http_client.h"
+
+#include <string>
+
+#include "gtest/gtest.h"
+
+namespace paimon::test {
+
+TEST(HttpClientUtilTest, ParseHttpHeaderLine) {
+    HttpHeaders headers;
+    const std::string line = "Content-Type: application/json\r\n";
+    ParseHttpHeaderLine(line.data(), line.size(), &headers);
+    ASSERT_EQ((HttpHeaders{{"content-type", "application/json"}}), headers);
+
+    const std::string overwrite = "content-type:  text/PLAIN \r\n";
+    ParseHttpHeaderLine(overwrite.data(), overwrite.size(), &headers);
+    ASSERT_EQ((HttpHeaders{{"content-type", "text/PLAIN"}}), headers);
+
+    // Lines without a colon (like the status line) and empty names are ignored.
+    const std::string status_line = "HTTP/1.1 200 OK\r\n";
+    ParseHttpHeaderLine(status_line.data(), status_line.size(), &headers);
+    const std::string empty_name = ": value\r\n";
+    ParseHttpHeaderLine(empty_name.data(), empty_name.size(), &headers);
+    ASSERT_EQ(1, headers.size());
+
+    const std::string empty_value = "x-empty:\r\n";
+    ParseHttpHeaderLine(empty_value.data(), empty_value.size(), &headers);
+    ASSERT_EQ("", headers.at("x-empty"));
+}
+
+}  // namespace paimon::test