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.

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

.github/workflows/build_and_test.yaml

@@ -67,6 +67,11 @@ jobs:
         uses: ./.github/actions/setup-ccache
         with:
           cache-key-prefix: ccache-${{ matrix.name }}
+      - name: Install dependencies
+        shell: bash
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y libcurl4-openssl-dev
       - name: Install Rust toolchain (tantivy-fts)
         if: ${{ !matrix.skip_rust }}
         shell: bash

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

CMakeLists.txt

@@ -63,9 +63,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

@@ -136,6 +136,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

@@ -147,6 +147,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/defs.h

@@ -112,6 +112,20 @@ struct PAIMON_EXPORT Options {
     /// Default value is local.
     static const char FILE_SYSTEM[];
 
+    /// "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[];
+
     /// "target-file-size" - Target size of a file. primary key table: the default value is 128 MB.
     /// append table: the default value is 256 MB.
     static const char TARGET_FILE_SIZE[];

src/paimon/CMakeLists.txt

@@ -395,6 +395,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}
@@ -408,6 +422,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
+               ${PAIMON_REST_LINK_LIBS}
                STATIC_LINK_LIBS
                arrow
                tbb
@@ -417,6 +432,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
+               ${PAIMON_REST_LINK_LIBS}
                SHARED_LINK_FLAGS
                ${PAIMON_VERSION_SCRIPT_FLAGS})
 
@@ -855,4 +871,18 @@ if(PAIMON_BUILD_TESTS)
         target_compile_definitions(paimon-fs-test PRIVATE PAIMON_ENABLE_NETWORK_TESTS)
     endif()
 
+    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/defs.cpp

@@ -31,6 +31,10 @@ const char Options::BUCKET[] = "bucket";
 const char Options::BUCKET_KEY[] = "bucket-key";
 const char Options::FILE_FORMAT[] = "file.format";
 const char Options::FILE_SYSTEM[] = "file-system";
+const char Options::METASTORE[] = "metastore";
+const char Options::URI[] = "uri";
+const char Options::TOKEN[] = "token";
+const char Options::TOKEN_PROVIDER[] = "token.provider";
 const char Options::TARGET_FILE_SIZE[] = "target-file-size";
 const char Options::BLOB_TARGET_FILE_SIZE[] = "blob.target-file-size";
 const char Options::BLOB_SPLIT_BY_FILE_SIZE[] = "blob.split-by-file-size";

src/paimon/core/catalog/catalog.cpp

@@ -18,8 +18,13 @@
 
 #include <utility>
 
+#include "paimon/common/utils/string_utils.h"
 #include "paimon/core/catalog/file_system_catalog.h"
 #include "paimon/core/core_options.h"
+#include "paimon/defs.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(Options::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);
 }

src/paimon/core/catalog/file_system_catalog.cpp

@@ -491,24 +491,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));
@@ -523,20 +505,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;
 }