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

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

- HttpClient: blocking libcurl client with exponential-backoff
  retries (429/503 retried for all methods, transport errors only
  for idempotent ones, honoring a numeric Retry-After header).
- RestApi: HTTP + JSON layer with "/v1/config" option merging
  (overrides > client options > defaults), "header." options as
  request headers, paged listing and http-status-to-Status error
  mapping including the x-request-id.
- RestCatalog: database/table operations and snapshot listing,
  "table-default." option defaults, system/branch table checks and
  schema conversion with the computed highestFieldId.
- Bear token authentication provider.
- CatalogOptions: catalog-level option keys (metastore, uri, token,
  token.provider) in a dedicated public header, mirroring the
  CatalogOptions/CoreOptions separation of the Java implementation.

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; CI enables it explicitly and installs the libcurl development
package.

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

Author: SteNicholas

CMakeLists.txt

@@ -66,9 +66,14 @@ 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)
+    find_package(CURL REQUIRED)
+    add_definitions(-DPAIMON_ENABLE_REST)
+endif()
 if(PAIMON_ENABLE_AVRO)
     add_definitions(-DPAIMON_ENABLE_AVRO)
 endif()

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

@@ -146,6 +146,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,43 @@
+/*
+ * 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 {
+
+/// Configuration options for Paimon catalogs.
+///
+/// The CatalogOptions struct contains static string constants that define catalog-level
+/// configuration keys. Table-level configuration 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.
+    static const char TOKEN_PROVIDER[];
+};
+
+}  // 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
@@ -403,6 +404,20 @@ set(PAIMON_CORE_SRCS
     core/utils/special_field_ids.cpp
     core/utils/tag_manager.cpp)
 
+set(PAIMON_REST_LINK_LIBS)
+if(PAIMON_ENABLE_REST)
+    list(APPEND
+         PAIMON_CORE_SRCS
+         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)
+    set(PAIMON_REST_LINK_LIBS CURL::libcurl)
+endif()
+
 add_paimon_lib(paimon
                SOURCES
                ${PAIMON_COMMON_SRCS}
@@ -417,6 +432,7 @@ add_paimon_lib(paimon
                Threads::Threads
                RapidJSON
                ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_REST_LINK_LIBS}
                STATIC_LINK_LIBS
                arrow
                tbb
@@ -427,6 +443,7 @@ add_paimon_lib(paimon
                Threads::Threads
                RapidJSON
                ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_REST_LINK_LIBS}
                SHARED_LINK_FLAGS
                ${PAIMON_VERSION_SCRIPT_FLAGS})
 
@@ -870,4 +887,18 @@ if(PAIMON_BUILD_TESTS)
                     EXTRA_INCLUDES
                     ${JINDOSDK_INCLUDE_DIR})
 
+    if(PAIMON_ENABLE_REST)
+        add_paimon_test(rest_test
+                        SOURCES
+                        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,26 @@
+/*
+ * 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";
+
+}  // namespace paimon

src/paimon/core/catalog/catalog.cpp

@@ -18,8 +18,13 @@
 
 #include <utility>
 
+#include "paimon/catalog_options.h"
+#include "paimon/common/utils/string_utils.h"
 #include "paimon/core/catalog/file_system_catalog.h"
 #include "paimon/core/core_options.h"
+#ifdef PAIMON_ENABLE_REST
+#include "paimon/rest/rest_catalog.h"
+#endif
 
 namespace paimon {
 
@@ -31,6 +36,22 @@ const char Catalog::DB_LOCATION_PROP[] = "location";
 Result<std::unique_ptr<Catalog>> Catalog::Create(const std::string& root_path,
                                                  const std::map<std::string, std::string>& options,
                                                  const std::shared_ptr<FileSystem>& file_system) {
+    std::string metastore = "filesystem";
+    auto metastore_iter = options.find(CatalogOptions::METASTORE);
+    if (metastore_iter != options.end()) {
+        metastore = StringUtils::ToLowerCase(metastore_iter->second);
+    }
+    if (metastore == "rest") {
+#ifdef PAIMON_ENABLE_REST
+        return RestCatalog::Create(root_path, options, file_system);
+#else
+        return Status::NotImplemented(
+            "the rest catalog requires building paimon with PAIMON_ENABLE_REST=ON");
+#endif
+    }
+    if (metastore != "filesystem") {
+        return Status::Invalid("unsupported metastore: ", metastore);
+    }
     PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CoreOptions::FromMap(options, file_system));
     return std::make_unique<FileSystemCatalog>(core_options.GetFileSystem(), root_path, options);
 }

src/paimon/core/catalog/file_system_catalog.cpp

@@ -521,24 +521,6 @@ Status FileSystemCatalog::RenameTable(const Identifier& from_table, const Identi
     return Status::OK();
 }
 
-namespace {
-SnapshotInfo::CommitKind ConvertCommitKind(Snapshot::CommitKind internal) {
-    if (internal == Snapshot::CommitKind::Append()) {
-        return SnapshotInfo::CommitKind::APPEND;
-    }
-    if (internal == Snapshot::CommitKind::Compact()) {
-        return SnapshotInfo::CommitKind::COMPACT;
-    }
-    if (internal == Snapshot::CommitKind::Overwrite()) {
-        return SnapshotInfo::CommitKind::OVERWRITE;
-    }
-    if (internal == Snapshot::CommitKind::Analyze()) {
-        return SnapshotInfo::CommitKind::ANALYZE;
-    }
-    return SnapshotInfo::CommitKind::UNKNOWN;
-}
-}  // namespace
-
 Result<std::vector<SnapshotInfo>> FileSystemCatalog::ListSnapshots(
     const Identifier& identifier, const std::string& branch) const {
     PAIMON_ASSIGN_OR_RAISE(bool exists, TableExists(identifier));
@@ -553,20 +535,9 @@ Result<std::vector<SnapshotInfo>> FileSystemCatalog::ListSnapshots(
 
     std::vector<SnapshotInfo> result;
     result.reserve(snapshots.size());
-
     for (const auto& snap : snapshots) {
-        SnapshotInfo info;
-        info.snapshot_id = snap.Id();
-        info.schema_id = snap.SchemaId();
-        info.commit_user = snap.CommitUser();
-        info.commit_kind = ConvertCommitKind(snap.GetCommitKind());
-        info.time_millis = snap.TimeMillis();
-        info.total_record_count = snap.TotalRecordCount();
-        info.delta_record_count = snap.DeltaRecordCount();
-        info.watermark = snap.Watermark();
-        result.push_back(std::move(info));
+        result.push_back(snap.ToSnapshotInfo());
     }
-
     return result;
 }
 

src/paimon/core/snapshot.cpp

@@ -313,4 +313,27 @@ Result<Snapshot> Snapshot::FromPath(const std::shared_ptr<FileSystem>& fs,
     return snapshot;
 }
 
+SnapshotInfo Snapshot::ToSnapshotInfo() const {
+    SnapshotInfo info;
+    info.snapshot_id = Id();
+    info.schema_id = SchemaId();
+    info.commit_user = CommitUser();
+    if (commit_kind_ == CommitKind::Append()) {
+        info.commit_kind = SnapshotInfo::CommitKind::APPEND;
+    } else if (commit_kind_ == CommitKind::Compact()) {
+        info.commit_kind = SnapshotInfo::CommitKind::COMPACT;
+    } else if (commit_kind_ == CommitKind::Overwrite()) {
+        info.commit_kind = SnapshotInfo::CommitKind::OVERWRITE;
+    } else if (commit_kind_ == CommitKind::Analyze()) {
+        info.commit_kind = SnapshotInfo::CommitKind::ANALYZE;
+    } else {
+        info.commit_kind = SnapshotInfo::CommitKind::UNKNOWN;
+    }
+    info.time_millis = TimeMillis();
+    info.total_record_count = TotalRecordCount();
+    info.delta_record_count = DeltaRecordCount();
+    info.watermark = Watermark();
+    return info;
+}
+
 }  // namespace paimon

src/paimon/core/snapshot.h

@@ -25,6 +25,7 @@
 
 #include "paimon/common/utils/jsonizable.h"
 #include "paimon/result.h"
+#include "paimon/snapshot/snapshot_info.h"
 #include "paimon/type_fwd.h"
 #include "rapidjson/allocators.h"
 #include "rapidjson/document.h"
@@ -129,6 +130,8 @@ class Snapshot : public Jsonizable<Snapshot> {
     bool operator==(const Snapshot& other) const;
     bool TEST_Equal(const Snapshot& other) const;
 
+    SnapshotInfo ToSnapshotInfo() const;
+
  public:
     static constexpr int64_t FIRST_SNAPSHOT_ID = 1;
     static constexpr int32_t TABLE_STORE_02_VERSION = 1;