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

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.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.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