diff --git a/.github/workflows/build_and_test.yaml b/.github/workflows/build_and_test.yaml index 3094cf72c..2fbf86a8d 100644 --- a/.github/workflows/build_and_test.yaml +++ b/.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 diff --git a/.github/workflows/gcc8_test.yaml b/.github/workflows/gcc8_test.yaml index c4316b032..245c62940 100644 --- a/.github/workflows/gcc8_test.yaml +++ b/.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 diff --git a/CMakeLists.txt b/CMakeLists.txt index 588abe489..422a83457 100644 --- a/CMakeLists.txt +++ b/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() diff --git a/ci/scripts/build_paimon.sh b/ci/scripts/build_paimon.sh index ba022423c..3e569f462 100755 --- a/ci/scripts/build_paimon.sh +++ b/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}" ) diff --git a/docs/source/building.rst b/docs/source/building.rst index e6014993b..2b065831c 100644 --- a/docs/source/building.rst +++ b/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 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/include/paimon/defs.h b/include/paimon/defs.h index cc128d478..a3e41737f 100644 --- a/include/paimon/defs.h +++ b/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[]; diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index d7cfda724..7425df98f 100644 --- a/src/paimon/CMakeLists.txt +++ b/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}) @@ -851,4 +867,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() diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp index 0df896a99..8d002ee82 100644 --- a/src/paimon/common/defs.cpp +++ b/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"; diff --git a/src/paimon/core/catalog/catalog.cpp b/src/paimon/core/catalog/catalog.cpp index 08b879c7a..2e1b00855 100644 --- a/src/paimon/core/catalog/catalog.cpp +++ b/src/paimon/core/catalog/catalog.cpp @@ -18,8 +18,13 @@ #include +#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> Catalog::Create(const std::string& root_path, const std::map& options, const std::shared_ptr& 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(core_options.GetFileSystem(), root_path); } diff --git a/src/paimon/core/catalog/file_system_catalog.cpp b/src/paimon/core/catalog/file_system_catalog.cpp index db9c2494e..efb0c6ebf 100644 --- a/src/paimon/core/catalog/file_system_catalog.cpp +++ b/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> FileSystemCatalog::ListSnapshots( const Identifier& identifier, const std::string& branch) const { PAIMON_ASSIGN_OR_RAISE(bool exists, TableExists(identifier)); @@ -523,20 +505,9 @@ Result> FileSystemCatalog::ListSnapshots( std::vector 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; } diff --git a/src/paimon/core/snapshot.cpp b/src/paimon/core/snapshot.cpp index 8e50a6464..f27846a12 100644 --- a/src/paimon/core/snapshot.cpp +++ b/src/paimon/core/snapshot.cpp @@ -313,4 +313,27 @@ Result Snapshot::FromPath(const std::shared_ptr& 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 diff --git a/src/paimon/core/snapshot.h b/src/paimon/core/snapshot.h index 2ef09e377..6520bf3ca 100644 --- a/src/paimon/core/snapshot.h +++ b/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 { 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; diff --git a/src/paimon/rest/http_client.cpp b/src/paimon/rest/http_client.cpp new file mode 100644 index 000000000..20f468a74 --- /dev/null +++ b/src/paimon/rest/http_client.cpp @@ -0,0 +1,265 @@ +/* + * 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/rest/http_client.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "curl/curl.h" +#include "fmt/format.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/rest/rest_util.h" + +namespace paimon { + +namespace { + +std::once_flag curl_global_init_flag; + +size_t WriteBodyCallback(char* data, size_t size, size_t nmemb, void* user_data) { + auto* body = static_cast(user_data); + body->append(data, size * nmemb); + return size * nmemb; +} + +// `StringUtils::Trim` only strips spaces; header lines end with "\r\n". +void TrimWhitespace(std::string* str) { + const char* whitespace = " \t\r\n"; + str->erase(str->find_last_not_of(whitespace) + 1); + str->erase(0, str->find_first_not_of(whitespace)); +} + +size_t WriteHeaderCallback(char* data, size_t size, size_t nmemb, void* user_data) { + auto* headers = static_cast*>(user_data); + std::string line(data, size * nmemb); + size_t colon = line.find(':'); + if (colon != std::string::npos) { + std::string name = line.substr(0, colon); + std::string value = line.substr(colon + 1); + TrimWhitespace(&name); + TrimWhitespace(&value); + if (!name.empty()) { + (*headers)[StringUtils::ToLowerCase(name)] = value; + } + } + return size * nmemb; +} + +bool IsIdempotent(const std::string& method) { + return method == "GET" || method == "HEAD" || method == "PUT" || method == "DELETE" || + method == "OPTIONS" || method == "TRACE"; +} + +bool IsRetriableCode(int64_t code) { + return code == 429 || code == 503; +} + +// Approximates the non-retriable exception classes of the Java +// `ExponentialHttpRequestRetryStrategy` (UnknownHostException, ConnectException, +// NoRouteToHostException, InterruptedIOException and SSLException): failures that do +// not become healthy by retrying. +bool IsRetriableTransportError(CURLcode code) { + switch (code) { + case CURLE_COULDNT_RESOLVE_PROXY: + case CURLE_COULDNT_RESOLVE_HOST: + case CURLE_COULDNT_CONNECT: + case CURLE_OPERATION_TIMEDOUT: + case CURLE_SSL_CONNECT_ERROR: + case CURLE_SSL_CERTPROBLEM: + case CURLE_SSL_CIPHER: + case CURLE_PEER_FAILED_VERIFICATION: + case CURLE_SSL_CACERT_BADFILE: + case CURLE_SSL_ISSUER_ERROR: + return false; + default: + return true; + } +} + +} // namespace + +HttpClient::HttpClient(const std::string& base_uri, const Config& config) + : base_uri_(base_uri), config_(config), logger_(Logger::GetLogger("HttpClient")) {} + +Result> HttpClient::Create(const std::string& base_uri) { + return Create(base_uri, Config()); +} + +Result> HttpClient::Create(const std::string& base_uri, + const Config& config) { + if (base_uri.empty()) { + return Status::Invalid("uri of the http client is empty"); + } + std::call_once(curl_global_init_flag, [] { curl_global_init(CURL_GLOBAL_DEFAULT); }); + return std::unique_ptr(new HttpClient(NormalizeUri(base_uri), config)); +} + +std::string HttpClient::NormalizeUri(const std::string& uri) { + std::string normalized = uri; + StringUtils::Trim(&normalized); + while (!normalized.empty() && normalized.back() == '/') { + normalized.pop_back(); + } + if (normalized.rfind("http://", 0) != 0 && normalized.rfind("https://", 0) != 0) { + normalized = "http://" + normalized; + } + return normalized; +} + +std::string HttpClient::BuildQueryString(const std::map& query_params) { + std::string query; + for (const auto& [key, value] : query_params) { + if (!query.empty()) { + query.push_back('&'); + } + query.append(RestUtil::EncodeString(key)); + query.push_back('='); + query.append(RestUtil::EncodeString(value)); + } + return query; +} + +Result HttpClient::ExecuteOnce( + const std::string& method, const std::string& url, + const std::map& headers, const std::string& body, + bool* transport_retriable) const { + CURL* curl = curl_easy_init(); + if (curl == nullptr) { + return Status::IOError("failed to create curl handle"); + } + Response response; + curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L); + curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, + static_cast(config_.connect_timeout_ms)); // NOLINT(runtime/int) + curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, + static_cast(config_.request_timeout_ms)); // NOLINT(runtime/int) + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteBodyCallback); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response.body); + curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, WriteHeaderCallback); + curl_easy_setopt(curl, CURLOPT_HEADERDATA, &response.headers); + + if (method == "POST") { + curl_easy_setopt(curl, CURLOPT_POST, 1L); + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.c_str()); + curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, + static_cast(body.size())); // NOLINT(runtime/int) + } else if (method == "DELETE") { + curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "DELETE"); + if (!body.empty()) { + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.c_str()); + curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, + static_cast(body.size())); // NOLINT(runtime/int) + } + } else if (method != "GET") { + curl_easy_cleanup(curl); + return Status::Invalid(fmt::format("unsupported http method: {}", method)); + } + + struct curl_slist* header_list = nullptr; + // Disable the automatic "Expect: 100-continue" behavior of libcurl on POST. + header_list = curl_slist_append(header_list, "Expect:"); + for (const auto& [name, value] : headers) { + header_list = curl_slist_append(header_list, (name + ": " + value).c_str()); + } + curl_easy_setopt(curl, CURLOPT_HTTPHEADER, header_list); + + CURLcode curl_code = curl_easy_perform(curl); + if (curl_code != CURLE_OK) { + curl_slist_free_all(header_list); + curl_easy_cleanup(curl); + *transport_retriable = IsRetriableTransportError(curl_code); + // Do not include the response body or full url in errors, they may carry + // sensitive information such as credentials. + return Status::IOError( + fmt::format("http {} request failed: {}", method, curl_easy_strerror(curl_code))); + } + // curl writes a `long` through the pointer, so the type cannot be narrowed here. + // NOLINTNEXTLINE(google-runtime-int) + long http_code = 0; // NOLINT(runtime/int) + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code); + response.code = http_code; + curl_slist_free_all(header_list); + curl_easy_cleanup(curl); + return response; +} + +int64_t HttpClient::GetRetryDelayMs(int32_t execution_count, const Response* response) const { + if (response != nullptr) { + auto iter = response->headers.find("retry-after"); + if (iter != response->headers.end()) { + std::optional retry_after_seconds = + StringUtils::StringToValue(iter->second); + if (retry_after_seconds && retry_after_seconds.value() >= 0) { + return retry_after_seconds.value() * 1000; + } + } + } + int64_t multiplier = static_cast(1) + << std::min(execution_count - 1, static_cast(6)); + int64_t delay_ms = config_.retry_base_delay_ms * multiplier; + if (delay_ms > 0) { + static thread_local std::mt19937 generator( + std::random_device{}()); // NOLINT(whitespace/braces) + std::uniform_int_distribution jitter(0, delay_ms / 10); + delay_ms += jitter(generator); + } + return delay_ms; +} + +Result HttpClient::Execute( + const std::string& method, const std::string& path, + const std::map& query_params, + const std::map& headers, const std::string& body) const { + std::string url = base_uri_ + path; + if (!query_params.empty()) { + url += "?" + BuildQueryString(query_params); + } + bool idempotent = IsIdempotent(method); + int32_t execution_count = 0; + while (true) { + execution_count++; + bool transport_retriable = false; + Result result = ExecuteOnce(method, url, headers, body, &transport_retriable); + bool retriable; + if (result.ok()) { + retriable = IsRetriableCode(result.value().code); + } else { + retriable = idempotent && transport_retriable; + } + if (!retriable || execution_count > config_.max_retries) { + return result; + } + int64_t delay_ms = + GetRetryDelayMs(execution_count, result.ok() ? &result.value() : nullptr); + std::string reason = result.ok() ? fmt::format("http status {}", result.value().code) + : result.status().ToString(); + PAIMON_LOG_WARN(logger_, "retrying %s %s in %lld ms (retry %d/%d): %s", method.c_str(), + path.c_str(), static_cast(delay_ms), // NOLINT(runtime/int) + execution_count, config_.max_retries, reason.c_str()); + if (delay_ms > 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms)); + } + } +} + +} // namespace paimon diff --git a/src/paimon/rest/http_client.h b/src/paimon/rest/http_client.h new file mode 100644 index 000000000..7be4113a1 --- /dev/null +++ b/src/paimon/rest/http_client.h @@ -0,0 +1,98 @@ +/* + * 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 +#include +#include +#include + +#include "paimon/logging.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +/// A blocking HTTP client for the REST catalog based on libcurl, mirroring the retry +/// behavior of `HttpClient` plus `ExponentialHttpRequestRetryStrategy` in the Java +/// implementation: HTTP 429/503 responses are retried for all methods, transport errors +/// are retried only for idempotent methods, with exponential backoff (honoring a numeric +/// `Retry-After` response header). +class HttpClient { + public: + struct Config { + int64_t connect_timeout_ms = 180 * 1000; + int64_t request_timeout_ms = 180 * 1000; + int32_t max_retries = 5; + int64_t retry_base_delay_ms = 1000; + }; + + struct Response { + int64_t code = 0; + std::string body; + /// Response headers with lower-cased names. + std::map headers; + + bool IsSuccessful() const { + return code == 200 || code == 202 || code == 204; + } + }; + + /// Creates a client against `base_uri`. The uri is normalized like the Java client: + /// trailing '/' stripped, "http://" prepended when no scheme is present. + static Result> Create(const std::string& base_uri); + static Result> Create(const std::string& base_uri, + const Config& config); + + /// Executes `method` ("GET", "POST" or "DELETE") on `path` (already url-encoded, + /// starting with '/'). Query parameter keys and values are url-encoded internally. + /// Returns the final response (which may carry a non-2xx code) or an error status + /// when the request could not be transported at all. Transport errors caused by + /// unresolvable/unreachable hosts, timeouts or TLS failures are never retried. + Result Execute(const std::string& method, const std::string& path, + const std::map& query_params, + const std::map& headers, + const std::string& body) const; + + const std::string& GetBaseUri() const { + return base_uri_; + } + + static std::string NormalizeUri(const std::string& uri); + + /// Builds "k1=v1&k2=v2" with url-encoded keys and values. + static std::string BuildQueryString(const std::map& query_params); + + private: + HttpClient(const std::string& base_uri, const Config& config); + + /// On a transport failure, `transport_retriable` reports whether the failure kind + /// may be retried (mirroring the non-retriable exception classes of the Java + /// strategy: unresolvable/unreachable hosts, timeouts and TLS failures are not). + Result ExecuteOnce(const std::string& method, const std::string& url, + const std::map& headers, + const std::string& body, bool* transport_retriable) const; + + /// Backoff before the retry following the `execution_count`-th attempt, in ms. + int64_t GetRetryDelayMs(int32_t execution_count, const Response* response) const; + + std::string base_uri_; + Config config_; + std::shared_ptr logger_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/http_client_test.cpp b/src/paimon/rest/http_client_test.cpp new file mode 100644 index 000000000..f2e919f39 --- /dev/null +++ b/src/paimon/rest/http_client_test.cpp @@ -0,0 +1,224 @@ +/* + * 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/rest/http_client.h" + +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/rest/mock_rest_server.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +namespace { +HttpClient::Config FastRetryConfig(int32_t max_retries) { + HttpClient::Config config; + config.max_retries = max_retries; + config.retry_base_delay_ms = 1; + return config; +} +} // namespace + +TEST(HttpClientTest, NormalizeUri) { + ASSERT_EQ("http://localhost:80", HttpClient::NormalizeUri("localhost:80/")); + ASSERT_EQ("http://localhost", HttpClient::NormalizeUri("http://localhost//")); + ASSERT_EQ("https://foo.bar", HttpClient::NormalizeUri(" https://foo.bar ")); +} + +TEST(HttpClientTest, BuildQueryString) { + ASSERT_EQ("a=1&b=x+y%2Fz", HttpClient::BuildQueryString({{"a", "1"}, {"b", "x y/z"}})); +} + +TEST(HttpClientTest, GetWithHeadersAndQuery) { + MockRestServer::Request last_request; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request& request) { + last_request = request; + MockRestServer::Response response; + response.body = R"({"ok": true})"; + return response; + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(server->GetBaseUri())); + ASSERT_OK_AND_ASSIGN(HttpClient::Response response, + client->Execute("GET", "/v1/config", {{"warehouse", "wh 1"}}, + {{"Authorization", "Bearer token1"}}, "")); + ASSERT_EQ(200, response.code); + ASSERT_EQ(R"({"ok": true})", response.body); + ASSERT_EQ("application/json", response.headers.at("content-type")); + ASSERT_EQ("GET", last_request.method); + ASSERT_EQ("/v1/config", last_request.path); + ASSERT_EQ("wh 1", last_request.query_params.at("warehouse")); + ASSERT_EQ("Bearer token1", last_request.headers.at("authorization")); +} + +TEST(HttpClientTest, PostBody) { + MockRestServer::Request last_request; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request& request) { + last_request = request; + return MockRestServer::Response(); + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(server->GetBaseUri())); + ASSERT_OK_AND_ASSIGN( + HttpClient::Response response, + client->Execute("POST", "/v1/databases", {}, {{"Content-Type", "application/json"}}, + R"({"name": "db1"})")); + ASSERT_EQ(200, response.code); + ASSERT_EQ("POST", last_request.method); + ASSERT_EQ(R"({"name": "db1"})", last_request.body); + ASSERT_EQ("application/json", last_request.headers.at("content-type")); +} + +TEST(HttpClientTest, NotFoundIsNotRetried) { + std::atomic request_count{0}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request& request) { + request_count++; + MockRestServer::Response response; + response.code = 404; + response.body = R"({"message": "not found", "code": 404})"; + return response; + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(server->GetBaseUri(), FastRetryConfig(5))); + ASSERT_OK_AND_ASSIGN(HttpClient::Response response, + client->Execute("GET", "/v1/databases/db1", {}, {}, "")); + ASSERT_EQ(404, response.code); + ASSERT_FALSE(response.IsSuccessful()); + ASSERT_EQ(1, request_count.load()); +} + +TEST(HttpClientTest, ServiceUnavailableIsRetried) { + std::atomic request_count{0}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request& request) { + MockRestServer::Response response; + if (request_count++ < 2) { + response.code = 503; + } else { + response.body = R"({"ok": true})"; + } + return response; + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(server->GetBaseUri(), FastRetryConfig(5))); + // 429/503 responses are retried even for non-idempotent POST. + ASSERT_OK_AND_ASSIGN(HttpClient::Response response, + client->Execute("POST", "/v1/databases", {}, {}, "{}")); + ASSERT_EQ(200, response.code); + ASSERT_EQ(3, request_count.load()); +} + +TEST(HttpClientTest, TooManyRequestsHonorsRetryAfter) { + std::atomic request_count{0}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request& request) { + MockRestServer::Response response; + if (request_count++ < 1) { + response.code = 429; + response.headers["Retry-After"] = "0"; + } + return response; + })); + // the backoff base is large so that the elapsed time below proves "Retry-After: 0" + // took precedence: without it this test would sleep for a minute + HttpClient::Config config; + config.max_retries = 5; + config.retry_base_delay_ms = 60 * 1000; + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(server->GetBaseUri(), config)); + auto start = std::chrono::steady_clock::now(); + ASSERT_OK_AND_ASSIGN(HttpClient::Response response, + client->Execute("GET", "/v1/databases", {}, {}, "")); + auto elapsed = std::chrono::steady_clock::now() - start; + ASSERT_EQ(200, response.code); + ASSERT_EQ(2, request_count.load()); + ASSERT_LT(elapsed, std::chrono::seconds(30)); +} + +TEST(HttpClientTest, RetriesExhausted) { + std::atomic request_count{0}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request& request) { + request_count++; + MockRestServer::Response response; + response.code = 503; + return response; + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(server->GetBaseUri(), FastRetryConfig(2))); + ASSERT_OK_AND_ASSIGN(HttpClient::Response response, + client->Execute("GET", "/v1/databases", {}, {}, "")); + ASSERT_EQ(503, response.code); + // initial attempt + 2 retries + ASSERT_EQ(3, request_count.load()); +} + +TEST(HttpClientTest, ConnectErrorFailsWithoutRetry) { + // Grab a port that is very likely closed by starting and stopping a server. A + // connect failure is not retried, mirroring the non-retriable ConnectException of + // the Java retry strategy. + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([](const MockRestServer::Request& request) { + return MockRestServer::Response(); + })); + std::string base_uri = server->GetBaseUri(); + server->Stop(); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(base_uri, FastRetryConfig(5))); + ASSERT_NOK(client->Execute("GET", "/v1/config", {}, {}, "").status()); +} + +TEST(HttpClientTest, EmptyReplyIsRetriedForIdempotentRequests) { + std::atomic request_count{0}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request& request) { + MockRestServer::Response response; + if (request_count++ < 1) { + // closing without a response is what Java reports as + // NoHttpResponseException, which is retriable + response.close_without_response = true; + } else { + response.body = R"({"ok": true})"; + } + return response; + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create(server->GetBaseUri(), FastRetryConfig(5))); + ASSERT_OK_AND_ASSIGN(HttpClient::Response response, + client->Execute("GET", "/v1/databases", {}, {}, "")); + ASSERT_EQ(200, response.code); + ASSERT_EQ(2, request_count.load()); + + // the same transport error is not retried for non-idempotent POST + request_count = 0; + ASSERT_NOK(client->Execute("POST", "/v1/databases", {}, {}, "{}").status()); + ASSERT_EQ(1, request_count.load()); +} + +TEST(HttpClientTest, UnsupportedMethod) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + HttpClient::Create("http://127.0.0.1:1")); + ASSERT_NOK_WITH_MSG(client->Execute("PATCH", "/", {}, {}, "").status(), + "unsupported http method"); +} + +} // namespace paimon::test diff --git a/src/paimon/rest/mock_rest_server.cpp b/src/paimon/rest/mock_rest_server.cpp new file mode 100644 index 000000000..ff2838345 --- /dev/null +++ b/src/paimon/rest/mock_rest_server.cpp @@ -0,0 +1,206 @@ +/* + * 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/rest/mock_rest_server.h" + +#include +#include +#include + +#include +#include +#include +#include + +#include "fmt/format.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/rest/rest_util.h" + +namespace paimon { + +namespace { + +bool ReceiveAll(int32_t fd, std::string* buffer, size_t min_size) { + char chunk[4096]; + while (buffer->size() < min_size) { + ssize_t received = ::recv(fd, chunk, sizeof(chunk), 0); + if (received <= 0) { + return false; + } + buffer->append(chunk, static_cast(received)); + } + return true; +} + +std::map ParseQuery(const std::string& query) { + std::map params; + for (const std::string& pair : StringUtils::Split(query, "&", /*ignore_empty=*/true)) { + size_t eq = pair.find('='); + if (eq == std::string::npos) { + params[RestUtil::DecodeString(pair)] = ""; + } else { + params[RestUtil::DecodeString(pair.substr(0, eq))] = + RestUtil::DecodeString(pair.substr(eq + 1)); + } + } + return params; +} + +} // namespace + +MockRestServer::MockRestServer(Handler handler, int32_t listen_fd, int32_t port) + : handler_(std::move(handler)), listen_fd_(listen_fd), port_(port) { + accept_thread_ = std::thread([this] { AcceptLoop(); }); +} + +Result> MockRestServer::Start(Handler handler) { + int32_t listen_fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (listen_fd < 0) { + return Status::IOError("mock rest server: failed to create socket: ", std::strerror(errno)); + } + int32_t reuse = 1; + ::setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); + struct sockaddr_in address; + std::memset(&address, 0, sizeof(address)); + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + address.sin_port = 0; + if (::bind(listen_fd, reinterpret_cast(&address), sizeof(address)) < 0) { + ::close(listen_fd); + return Status::IOError("mock rest server: failed to bind: ", std::strerror(errno)); + } + if (::listen(listen_fd, 16) < 0) { + ::close(listen_fd); + return Status::IOError("mock rest server: failed to listen: ", std::strerror(errno)); + } + socklen_t address_len = sizeof(address); + if (::getsockname(listen_fd, reinterpret_cast(&address), &address_len) < 0) { + ::close(listen_fd); + return Status::IOError("mock rest server: failed to get port: ", std::strerror(errno)); + } + int32_t port = ntohs(address.sin_port); + return std::unique_ptr(new MockRestServer(std::move(handler), listen_fd, port)); +} + +MockRestServer::~MockRestServer() { + Stop(); +} + +void MockRestServer::Stop() { + if (stopped_.exchange(true)) { + return; + } + ::shutdown(listen_fd_, SHUT_RDWR); + ::close(listen_fd_); + if (accept_thread_.joinable()) { + accept_thread_.join(); + } +} + +std::string MockRestServer::GetBaseUri() const { + return fmt::format("http://127.0.0.1:{}", port_); +} + +void MockRestServer::AcceptLoop() { + while (!stopped_.load()) { + int32_t connection_fd = ::accept(listen_fd_, nullptr, nullptr); + if (connection_fd < 0) { + if (stopped_.load()) { + return; + } + continue; + } + HandleConnection(connection_fd); + ::close(connection_fd); + } +} + +void MockRestServer::HandleConnection(int32_t connection_fd) { + std::string buffer; + size_t header_end; + while ((header_end = buffer.find("\r\n\r\n")) == std::string::npos) { + if (!ReceiveAll(connection_fd, &buffer, buffer.size() + 1)) { + return; + } + } + + Request request; + std::string header_part = buffer.substr(0, header_end); + std::vector lines = StringUtils::Split(header_part, "\r\n", + /*ignore_empty=*/true); + if (lines.empty()) { + return; + } + std::vector request_line = StringUtils::Split(lines[0], " ", + /*ignore_empty=*/true); + if (request_line.size() < 2) { + return; + } + request.method = request_line[0]; + std::string target = request_line[1]; + size_t question = target.find('?'); + if (question == std::string::npos) { + request.path = RestUtil::DecodeString(target); + } else { + request.path = RestUtil::DecodeString(target.substr(0, question)); + request.query_params = ParseQuery(target.substr(question + 1)); + } + size_t content_length = 0; + for (size_t i = 1; i < lines.size(); i++) { + size_t colon = lines[i].find(':'); + if (colon == std::string::npos) { + continue; + } + std::string name = lines[i].substr(0, colon); + std::string value = lines[i].substr(colon + 1); + StringUtils::Trim(&name); + StringUtils::Trim(&value); + request.headers[StringUtils::ToLowerCase(name)] = value; + } + auto length_iter = request.headers.find("content-length"); + if (length_iter != request.headers.end()) { + content_length = + static_cast(std::strtoul(length_iter->second.c_str(), nullptr, 10)); + } + size_t body_begin = header_end + 4; + if (!ReceiveAll(connection_fd, &buffer, body_begin + content_length)) { + return; + } + request.body = buffer.substr(body_begin, content_length); + + Response response = handler_(request); + if (response.close_without_response) { + return; + } + std::string extra_headers; + for (const auto& [name, value] : response.headers) { + extra_headers += fmt::format("{}: {}\r\n", name, value); + } + std::string payload = fmt::format( + "HTTP/1.1 {} MOCK\r\nContent-Type: {}\r\nContent-Length: {}\r\n{}Connection: " + "close\r\n\r\n{}", + response.code, response.content_type, response.body.size(), extra_headers, response.body); + size_t sent = 0; + while (sent < payload.size()) { + ssize_t written = ::send(connection_fd, payload.data() + sent, payload.size() - sent, 0); + if (written <= 0) { + return; + } + sent += static_cast(written); + } +} + +} // namespace paimon diff --git a/src/paimon/rest/mock_rest_server.h b/src/paimon/rest/mock_rest_server.h new file mode 100644 index 000000000..ad723cbcc --- /dev/null +++ b/src/paimon/rest/mock_rest_server.h @@ -0,0 +1,88 @@ +/* + * 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 +#include +#include +#include +#include +#include + +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +/// A minimal blocking HTTP/1.1 server for REST catalog unit tests. It listens on a +/// random port of 127.0.0.1, parses one request per connection and answers it with the +/// response returned by the registered handler. Test only, not production code. +class MockRestServer { + public: + struct Request { + std::string method; + /// Url-decoded path, e.g. "/v1/config". + std::string path; + /// Url-decoded query parameters. + std::map query_params; + /// Headers with lower-cased names. + std::map headers; + std::string body; + }; + + struct Response { + int32_t code = 200; + std::string body; + std::string content_type = "application/json"; + /// Extra response headers, e.g. "Retry-After". + std::map headers; + /// Close the connection without sending any response, to simulate a retriable + /// transport error on the client. + bool close_without_response = false; + }; + + using Handler = std::function; + + /// Starts the server with `handler` invoked for every request (on the server + /// thread). + static Result> Start(Handler handler); + + ~MockRestServer(); + + void Stop(); + + int32_t GetPort() const { + return port_; + } + + /// "http://127.0.0.1:" + std::string GetBaseUri() const; + + private: + MockRestServer(Handler handler, int32_t listen_fd, int32_t port); + + void AcceptLoop(); + void HandleConnection(int32_t connection_fd); + + Handler handler_; + int32_t listen_fd_ = -1; + int32_t port_ = 0; + std::atomic stopped_{false}; + std::thread accept_thread_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/resource_paths.cpp b/src/paimon/rest/resource_paths.cpp new file mode 100644 index 000000000..08c1b0d65 --- /dev/null +++ b/src/paimon/rest/resource_paths.cpp @@ -0,0 +1,64 @@ +/* + * 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/rest/resource_paths.h" + +#include "paimon/rest/rest_util.h" + +namespace paimon { + +namespace { +constexpr const char kV1[] = "/v1"; +} // namespace + +ResourcePaths::ResourcePaths(const std::string& prefix) { + base_ = kV1; + if (!prefix.empty()) { + base_ += "/" + RestUtil::EncodeString(prefix); + } +} + +std::string ResourcePaths::Config() { + return std::string(kV1) + "/config"; +} + +std::string ResourcePaths::Databases() const { + return base_ + "/databases"; +} + +std::string ResourcePaths::Database(const std::string& database_name) const { + return Databases() + "/" + RestUtil::EncodeString(database_name); +} + +std::string ResourcePaths::Tables(const std::string& database_name) const { + return Database(database_name) + "/tables"; +} + +std::string ResourcePaths::Table(const std::string& database_name, + const std::string& table_name) const { + return Tables(database_name) + "/" + RestUtil::EncodeString(table_name); +} + +std::string ResourcePaths::RenameTable() const { + return base_ + "/tables/rename"; +} + +std::string ResourcePaths::Snapshots(const std::string& database_name, + const std::string& table_name) const { + return Table(database_name, table_name) + "/snapshots"; +} + +} // namespace paimon diff --git a/src/paimon/rest/resource_paths.h b/src/paimon/rest/resource_paths.h new file mode 100644 index 000000000..91d343aee --- /dev/null +++ b/src/paimon/rest/resource_paths.h @@ -0,0 +1,45 @@ +/* + * 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 + +namespace paimon { + +/// Builds resource paths of the REST catalog server, mirroring `ResourcePaths` in the +/// Java implementation. All path segments are url-encoded; `prefix` (usually pushed down +/// by the server through `/v1/config`) may be empty, in which case it is skipped. +class ResourcePaths { + public: + explicit ResourcePaths(const std::string& prefix); + + /// "/v1/config", the only path that does not carry the prefix. + static std::string Config(); + + std::string Databases() const; + std::string Database(const std::string& database_name) const; + std::string Tables(const std::string& database_name) const; + std::string Table(const std::string& database_name, const std::string& table_name) const; + std::string RenameTable() const; + std::string Snapshots(const std::string& database_name, const std::string& table_name) const; + + private: + /// "/v1" or "/v1/{encoded prefix}". + std::string base_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/rest_api.cpp b/src/paimon/rest/rest_api.cpp new file mode 100644 index 000000000..5b3af0e2d --- /dev/null +++ b/src/paimon/rest/rest_api.cpp @@ -0,0 +1,243 @@ +/* + * 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/rest/rest_api.h" + +#include + +#include "fmt/format.h" +#include "paimon/common/utils/rapidjson_util.h" +#include "paimon/defs.h" +#include "paimon/rest/rest_util.h" + +namespace paimon { + +namespace { +// The `PAIMON_ASSIGN_OR_RAISE` macro cannot take a declaration type containing a comma. +using StringMap = std::map; +} // namespace + +RestApi::RestApi(std::unique_ptr client, std::shared_ptr auth_provider, + const std::map& base_headers, + const std::map& options, const ResourcePaths& paths) + : client_(std::move(client)), + auth_provider_(std::move(auth_provider)), + base_headers_(base_headers), + options_(options), + resource_paths_(paths) {} + +Result> RestApi::Create(const std::map& options, + const std::string& warehouse, bool config_required, + const HttpClient::Config& http_config) { + auto uri_iter = options.find(Options::URI); + if (uri_iter == options.end() || uri_iter->second.empty()) { + return Status::Invalid( + fmt::format("option '{}' must be configured for the rest catalog", Options::URI)); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr client, + HttpClient::Create(uri_iter->second, http_config)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr auth_provider, + AuthProvider::Create(options)); + + std::map merged_options = options; + std::map base_headers = + RestUtil::ExtractPrefixMap(options, kHeaderOptionPrefix); + if (config_required) { + std::map query_params; + if (!warehouse.empty()) { + query_params[kQueryParamWarehouse] = warehouse; + } + RestAuthParameter auth_parameter; + auth_parameter.method = "GET"; + auth_parameter.resource_path = ResourcePaths::Config(); + for (const auto& [key, value] : query_params) { + auth_parameter.parameters[key] = RestUtil::EncodeString(value); + } + PAIMON_ASSIGN_OR_RAISE(StringMap headers, + auth_provider->MergeAuthHeader(base_headers, auth_parameter)); + PAIMON_ASSIGN_OR_RAISE( + HttpClient::Response response, + client->Execute("GET", ResourcePaths::Config(), query_params, headers, "")); + if (!response.IsSuccessful()) { + return ErrorToStatus(response); + } + ConfigResponse config; + PAIMON_RETURN_NOT_OK(RapidJsonUtil::FromJsonString(response.body, &config)); + merged_options = config.Merge(options); + for (const auto& [key, value] : + RestUtil::ExtractPrefixMap(merged_options, kHeaderOptionPrefix)) { + base_headers[key] = value; + } + } + std::string prefix; + auto prefix_iter = merged_options.find(kOptionPrefix); + if (prefix_iter != merged_options.end()) { + prefix = prefix_iter->second; + } + return std::unique_ptr(new RestApi(std::move(client), std::move(auth_provider), + base_headers, merged_options, + ResourcePaths(prefix))); +} + +Status RestApi::ErrorToStatus(const HttpClient::Response& response) { + std::string message; + if (!response.body.empty()) { + ErrorResponse error("", "", "", 0); + if (RapidJsonUtil::FromJsonString(response.body, &error).ok()) { + message = error.GetMessage(); + if (!error.GetResourceType().empty()) { + message += fmt::format(" (resource type: {}, resource name: {})", + error.GetResourceType(), error.GetResourceName()); + } + } + } + if (message.empty()) { + message = fmt::format("rest server returned http status {}", response.code); + } + // Mirrors DefaultErrorHandler: carry the request id through, it is what the server + // operators need to trace a failed call. + auto request_id_iter = response.headers.find("x-request-id"); + if (request_id_iter != response.headers.end() && !request_id_iter->second.empty() && + request_id_iter->second != "unknown") { + message += fmt::format(" requestId:{}", request_id_iter->second); + } + switch (response.code) { + case 400: + return Status::Invalid(message); + case 401: + return Status::IOError("not authorized: ", message); + case 403: + return Status::IOError("forbidden: ", message); + case 404: + return Status::NotExist(message); + case 409: + return Status::Exist(message); + case 500: + return Status::IOError("server error: ", message); + case 501: + return Status::NotImplemented(message); + case 503: + return Status::IOError("service unavailable: ", message); + default: + return Status::IOError( + fmt::format("rest request failed with code {}: {}", response.code, message)); + } +} + +Result RestApi::Execute( + const std::string& method, const std::string& path, + const std::map& query_params, const std::string& body) const { + RestAuthParameter auth_parameter; + auth_parameter.method = method; + auth_parameter.resource_path = path; + auth_parameter.data = body; + for (const auto& [key, value] : query_params) { + auth_parameter.parameters[key] = RestUtil::EncodeString(value); + } + PAIMON_ASSIGN_OR_RAISE(StringMap headers, + auth_provider_->MergeAuthHeader(base_headers_, auth_parameter)); + if (!body.empty()) { + headers["Content-Type"] = "application/json"; + } + PAIMON_ASSIGN_OR_RAISE(HttpClient::Response response, + client_->Execute(method, path, query_params, headers, body)); + if (!response.IsSuccessful()) { + return ErrorToStatus(response); + } + return response; +} + +template +Result RestApi::GetEntity(const std::string& path, + const std::map& query_params) const { + PAIMON_ASSIGN_OR_RAISE(HttpClient::Response response, Execute("GET", path, query_params, "")); + ResponseT entity; + PAIMON_RETURN_NOT_OK(RapidJsonUtil::FromJsonString(response.body, &entity)); + return entity; +} + +template +Result> RestApi::ListAllPages( + const std::string& path) const { + std::vector items; + std::map query_params; + while (true) { + PAIMON_ASSIGN_OR_RAISE(ResponseT response, GetEntity(path, query_params)); + const auto& data = response.Data(); + items.insert(items.end(), data.begin(), data.end()); + const std::optional& next_page_token = response.NextPageToken(); + if (!next_page_token || next_page_token.value().empty() || data.empty()) { + return items; + } + query_params[kQueryParamPageToken] = next_page_token.value(); + } +} + +Result> RestApi::ListDatabases() const { + return ListAllPages(resource_paths_.Databases()); +} + +Status RestApi::CreateDatabase(const std::string& name, + const std::map& options) const { + CreateDatabaseRequest request(name, options); + PAIMON_ASSIGN_OR_RAISE(std::string body, request.ToJsonString()); + return Execute("POST", resource_paths_.Databases(), {}, body).status(); +} + +Result RestApi::GetDatabase(const std::string& name) const { + return GetEntity(resource_paths_.Database(name), {}); +} + +Status RestApi::DropDatabase(const std::string& name) const { + return Execute("DELETE", resource_paths_.Database(name), {}, "").status(); +} + +Result> RestApi::ListTables(const std::string& database_name) const { + return ListAllPages(resource_paths_.Tables(database_name)); +} + +Result RestApi::GetTable(const Identifier& identifier) const { + return GetEntity( + resource_paths_.Table(identifier.GetDatabaseName(), identifier.GetTableName()), {}); +} + +Status RestApi::CreateTable(const Identifier& identifier, const std::string& schema_json) const { + CreateTableRequest request(identifier.GetDatabaseName(), identifier.GetTableName(), + schema_json); + PAIMON_ASSIGN_OR_RAISE(std::string body, request.ToJsonString()); + return Execute("POST", resource_paths_.Tables(identifier.GetDatabaseName()), {}, body).status(); +} + +Status RestApi::DropTable(const Identifier& identifier) const { + return Execute("DELETE", + resource_paths_.Table(identifier.GetDatabaseName(), identifier.GetTableName()), + {}, "") + .status(); +} + +Status RestApi::RenameTable(const Identifier& from_table, const Identifier& to_table) const { + RenameTableRequest request(from_table.GetDatabaseName(), from_table.GetTableName(), + to_table.GetDatabaseName(), to_table.GetTableName()); + PAIMON_ASSIGN_OR_RAISE(std::string body, request.ToJsonString()); + return Execute("POST", resource_paths_.RenameTable(), {}, body).status(); +} + +Result> RestApi::ListSnapshots(const Identifier& identifier) const { + return ListAllPages( + resource_paths_.Snapshots(identifier.GetDatabaseName(), identifier.GetTableName())); +} + +} // namespace paimon diff --git a/src/paimon/rest/rest_api.h b/src/paimon/rest/rest_api.h new file mode 100644 index 000000000..76841865a --- /dev/null +++ b/src/paimon/rest/rest_api.h @@ -0,0 +1,111 @@ +/* + * 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 +#include +#include +#include + +#include "paimon/catalog/identifier.h" +#include "paimon/core/snapshot.h" +#include "paimon/rest/http_client.h" +#include "paimon/rest/resource_paths.h" +#include "paimon/rest/rest_auth.h" +#include "paimon/rest/rest_messages.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +/// The client of the REST catalog server, mirroring `RESTApi` in the Java +/// implementation. This layer only talks HTTP + JSON and never touches the file system. +class RestApi { + public: + static constexpr const char* kQueryParamPageToken = "pageToken"; + static constexpr const char* kQueryParamWarehouse = "warehouse"; + /// Option key of the url prefix, usually pushed down by "/v1/config". + static constexpr const char* kOptionPrefix = "prefix"; + /// Options with this prefix are sent as http headers (with the prefix stripped). + static constexpr const char* kHeaderOptionPrefix = "header."; + + /// Creates the api client. + /// + /// @param options Client side options; `Options::URI` and `Options::TOKEN_PROVIDER` + /// are required. + /// @param warehouse Warehouse sent as query parameter of "/v1/config"; may be empty. + /// @param config_required When true, fetch "/v1/config" and merge the server + /// defaults/overrides into `options` with the precedence + /// overrides > client options > defaults. + /// @param http_config Transport level settings, mainly overridable for tests. + static Result> Create( + const std::map& options, const std::string& warehouse, + bool config_required, const HttpClient::Config& http_config = HttpClient::Config()); + + /// Options merged with the server side config. + const std::map& GetMergedOptions() const { + return options_; + } + + Result> ListDatabases() const; + Status CreateDatabase(const std::string& name, + const std::map& options) const; + Result GetDatabase(const std::string& name) const; + Status DropDatabase(const std::string& name) const; + + Result> ListTables(const std::string& database_name) const; + Result GetTable(const Identifier& identifier) const; + /// `schema_json` uses the Java `org.apache.paimon.schema.Schema` JSON layout. + Status CreateTable(const Identifier& identifier, const std::string& schema_json) const; + Status DropTable(const Identifier& identifier) const; + Status RenameTable(const Identifier& from_table, const Identifier& to_table) const; + + Result> ListSnapshots(const Identifier& identifier) const; + + /// Maps a non-successful http response to a status, mirroring `DefaultErrorHandler`: + /// 404 becomes `NotExist`, 409 becomes `Exist`, 400 becomes `Invalid`, 501 becomes + /// `NotImplemented` and the other codes become `IOError`. + static Status ErrorToStatus(const HttpClient::Response& response); + + private: + RestApi(std::unique_ptr client, std::shared_ptr auth_provider, + const std::map& base_headers, + const std::map& options, const ResourcePaths& paths); + + /// Executes one request with authentication headers and returns the response when it + /// is successful, otherwise the mapped error status. + Result Execute(const std::string& method, const std::string& path, + const std::map& query_params, + const std::string& body) const; + + template + Result GetEntity(const std::string& path, + const std::map& query_params) const; + + /// Fetches all pages of a paged listing api, mirroring `listDataFromPageApi`. + template + Result> ListAllPages(const std::string& path) const; + + std::unique_ptr client_; + std::shared_ptr auth_provider_; + /// Base http headers extracted from the "header." options. + std::map base_headers_; + std::map options_; + ResourcePaths resource_paths_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/rest_auth.cpp b/src/paimon/rest/rest_auth.cpp new file mode 100644 index 000000000..d262f5222 --- /dev/null +++ b/src/paimon/rest/rest_auth.cpp @@ -0,0 +1,53 @@ +/* + * 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/rest/rest_auth.h" + +#include "fmt/format.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/defs.h" + +namespace paimon { + +Result> BearTokenAuthProvider::MergeAuthHeader( + const std::map& base_header, + const RestAuthParameter& parameter) const { + std::map headers = base_header; + headers["Authorization"] = "Bearer " + token_; + return headers; +} + +Result> AuthProvider::Create( + const std::map& options) { + auto provider_iter = options.find(Options::TOKEN_PROVIDER); + if (provider_iter == options.end() || provider_iter->second.empty()) { + return Status::Invalid(fmt::format("option '{}' must be configured for the rest catalog", + Options::TOKEN_PROVIDER)); + } + std::string provider = StringUtils::ToLowerCase(provider_iter->second); + if (provider == "bear") { + auto token_iter = options.find(Options::TOKEN); + if (token_iter == options.end() || token_iter->second.empty()) { + return Status::Invalid(fmt::format( + "option '{}' must be configured for the bear token provider", Options::TOKEN)); + } + return std::make_shared(token_iter->second); + } + return Status::NotImplemented( + fmt::format("unsupported token provider: {}, only 'bear' is supported for now", provider)); +} + +} // namespace paimon diff --git a/src/paimon/rest/rest_auth.h b/src/paimon/rest/rest_auth.h new file mode 100644 index 000000000..f4eadbea2 --- /dev/null +++ b/src/paimon/rest/rest_auth.h @@ -0,0 +1,69 @@ +/* + * 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 +#include +#include + +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +/// Input of one request signature, mirroring `RESTAuthParameter` in the Java +/// implementation. +struct RestAuthParameter { + std::string method; + std::string resource_path; + /// Query parameters with url-encoded values. + std::map parameters; + /// Request body, empty when the request carries none. + std::string data; +}; + +/// Generates authentication headers for REST catalog requests, mirroring `AuthProvider` +/// in the Java implementation. +class AuthProvider { + public: + virtual ~AuthProvider() = default; + + /// Returns `base_header` merged with the authentication headers of this provider. + virtual Result> MergeAuthHeader( + const std::map& base_header, + const RestAuthParameter& parameter) const = 0; + + /// Creates the provider configured by `Options::TOKEN_PROVIDER`. + static Result> Create( + const std::map& options); +}; + +/// Adds `Authorization: Bearer `, mirroring `BearTokenAuthProvider` in the Java +/// implementation. +class BearTokenAuthProvider : public AuthProvider { + public: + explicit BearTokenAuthProvider(const std::string& token) : token_(token) {} + + Result> MergeAuthHeader( + const std::map& base_header, + const RestAuthParameter& parameter) const override; + + private: + std::string token_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/rest_catalog.cpp b/src/paimon/rest/rest_catalog.cpp new file mode 100644 index 000000000..b936f179e --- /dev/null +++ b/src/paimon/rest/rest_catalog.cpp @@ -0,0 +1,447 @@ +/* + * 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/rest/rest_catalog.h" + +#include +#include +#include +#include + +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "paimon/catalog/table.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/rapidjson_util.h" +#include "paimon/core/core_options.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/system/system_table.h" +#include "paimon/core/table/system/system_table_schema.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/rest/rest_util.h" +#include "rapidjson/document.h" + +namespace arrow { +class Schema; +} // namespace arrow + +namespace paimon { + +namespace { + +/// Mirrors `SpecialFields.SYSTEM_FIELD_ID_START` in the Java implementation; system +/// fields are excluded when computing the highest field id. +constexpr int32_t kSystemFieldIdStart = std::numeric_limits::max() / 2; + +/// The option key of the table path, mirrors `CoreOptions.PATH` in the Java +/// implementation. +constexpr const char kPathOption[] = "path"; + +/// Mirrors `Catalog.TABLE_DEFAULT_OPTION_PREFIX` in the Java implementation. +constexpr const char kTableDefaultOptionPrefix[] = "table-default."; + +void CollectFieldIds(const rapidjson::Value& fields, int32_t* max_id); + +void CollectTypeFieldIds(const rapidjson::Value& type, int32_t* max_id) { + if (!type.IsObject()) { + return; + } + // ROW type carries nested "fields", ARRAY/MULTISET carry "element" and MAP carries + // "key"/"value". + if (type.HasMember("fields") && type["fields"].IsArray()) { + CollectFieldIds(type["fields"], max_id); + } + if (type.HasMember("element")) { + CollectTypeFieldIds(type["element"], max_id); + } + if (type.HasMember("key")) { + CollectTypeFieldIds(type["key"], max_id); + } + if (type.HasMember("value")) { + CollectTypeFieldIds(type["value"], max_id); + } +} + +void CollectFieldIds(const rapidjson::Value& fields, int32_t* max_id) { + for (const auto& field : fields.GetArray()) { + if (!field.IsObject()) { + continue; + } + if (field.HasMember("id") && field["id"].IsInt()) { + int32_t id = field["id"].GetInt(); + if (id < kSystemFieldIdStart) { + *max_id = std::max(*max_id, id); + } + } + if (field.HasMember("type")) { + CollectTypeFieldIds(field["type"], max_id); + } + } +} + +/// Mirrors `RowType.currentHighestFieldId` in the Java implementation. +int32_t ComputeHighestFieldId(const rapidjson::Value& fields) { + int32_t max_id = -1; + CollectFieldIds(fields, &max_id); + return max_id; +} + +} // namespace + +RestCatalog::RestCatalog(std::unique_ptr api, const std::shared_ptr& fs, + const std::string& warehouse) + : api_(std::move(api)), + fs_(fs), + warehouse_(warehouse), + table_default_options_( + RestUtil::ExtractPrefixMap(api_->GetMergedOptions(), kTableDefaultOptionPrefix)), + logger_(Logger::GetLogger("RestCatalog")) {} + +Result> RestCatalog::Create( + const std::string& warehouse, const std::map& options, + const std::shared_ptr& file_system, const HttpClient::Config& http_config) { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr api, + RestApi::Create(options, warehouse, /*config_required=*/true, http_config)); + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, + CoreOptions::FromMap(api->GetMergedOptions(), file_system)); + return std::unique_ptr( + new RestCatalog(std::move(api), core_options.GetFileSystem(), warehouse)); +} + +const std::map& RestCatalog::GetMergedOptions() const { + return api_->GetMergedOptions(); +} + +bool RestCatalog::IsSystemDatabase(const std::string& db_name) { + return db_name == SYSTEM_DATABASE_NAME; +} + +Status RestCatalog::CheckNotSystemTable(const Identifier& identifier, const std::string& action) { + if (IsSystemDatabase(identifier.GetDatabaseName())) { + return Status::Invalid( + fmt::format("Cannot {} for system table {}.", action, identifier.ToString())); + } + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + return Status::Invalid( + fmt::format("Cannot {} for system table {}.", action, identifier.ToString())); + } + return Status::OK(); +} + +Status RestCatalog::CheckNotBranch(const Identifier& identifier, const std::string& action) { + PAIMON_ASSIGN_OR_RAISE(std::optional branch, identifier.GetBranchName()); + if (branch) { + return Status::Invalid( + fmt::format("Cannot {} for branch table {}.", action, identifier.ToString())); + } + return Status::OK(); +} + +Status RestCatalog::CreateDatabase(const std::string& name, + const std::map& options, + bool ignore_if_exists) { + if (IsSystemDatabase(name)) { + return Status::Invalid(fmt::format("Cannot create database for system database {}.", name)); + } + Status status = api_->CreateDatabase(name, options); + if (status.IsExist() && ignore_if_exists) { + return Status::OK(); + } + return status; +} + +Result> RestCatalog::ListDatabases() const { + return api_->ListDatabases(); +} + +Result RestCatalog::DatabaseExists(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented( + "do not support checking DatabaseExists for system database."); + } + Result response = api_->GetDatabase(db_name); + if (response.ok()) { + return true; + } + if (response.status().IsNotExist()) { + return false; + } + return response.status(); +} + +Status RestCatalog::DropDatabase(const std::string& name, bool ignore_if_not_exists, bool cascade) { + if (IsSystemDatabase(name)) { + return Status::Invalid(fmt::format("Cannot drop system database {}.", name)); + } + if (!cascade) { + Result> tables = ListTables(name); + if (!tables.ok()) { + if (tables.status().IsNotExist() && ignore_if_not_exists) { + return Status::OK(); + } + return tables.status(); + } + if (!tables.value().empty()) { + return Status::Invalid( + fmt::format("Cannot drop non-empty database {}. Use cascade=true to force.", name)); + } + } + Status status = api_->DropDatabase(name); + if (status.IsNotExist() && ignore_if_not_exists) { + return Status::OK(); + } + return status; +} + +std::string RestCatalog::GetDatabaseLocation(const std::string& db_name) const { + Result response = api_->GetDatabase(db_name); + if (!response.ok()) { + PAIMON_LOG_WARN(logger_, "failed to get location of database %s: %s", db_name.c_str(), + response.status().ToString().c_str()); + return ""; + } + return response.value().GetLocation(); +} + +Result> RestCatalog::ListTables(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented("do not support listing tables for system database."); + } + return api_->ListTables(db_name); +} + +Status RestCatalog::CreateTable(const Identifier& identifier, ArrowSchema* c_schema, + const std::vector& partition_keys, + const std::vector& primary_keys, + const std::map& options, + bool ignore_if_exists) { + PAIMON_RETURN_NOT_OK(CheckNotSystemTable(identifier, "create table")); + PAIMON_RETURN_NOT_OK(CheckNotBranch(identifier, "create table")); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr schema, + arrow::ImportSchema(c_schema)); + std::map effective_options = options; + for (const auto& [key, value] : table_default_options_) { + effective_options.emplace(key, value); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_schema, + TableSchema::Create(TableSchema::FIRST_SCHEMA_ID, schema, partition_keys, + primary_keys, effective_options)); + std::string schema_json; + try { + rapidjson::Document doc; + doc.SetObject(); + rapidjson::Document::AllocatorType& allocator = doc.GetAllocator(); + doc.AddMember(rapidjson::StringRef("fields"), + RapidJsonUtil::SerializeValue(table_schema->Fields(), &allocator).Move(), + allocator); + doc.AddMember(rapidjson::StringRef("partitionKeys"), + RapidJsonUtil::SerializeValue(partition_keys, &allocator).Move(), allocator); + doc.AddMember(rapidjson::StringRef("primaryKeys"), + RapidJsonUtil::SerializeValue(primary_keys, &allocator).Move(), allocator); + doc.AddMember(rapidjson::StringRef("options"), + RapidJsonUtil::SerializeValue(effective_options, &allocator).Move(), + allocator); + schema_json = RestUtil::JsonToString(doc); + } catch (const std::exception& e) { + return Status::SerializationError("failed to serialize create table schema: ", e.what()); + } + Status status = api_->CreateTable(identifier, schema_json); + if (status.IsExist() && ignore_if_exists) { + return Status::OK(); + } + return status; +} + +Result RestCatalog::TableExists(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::optional system_table_name, + identifier.GetSystemTableName()); + if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) { + return false; + } + } + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + Identifier data_identifier(identifier.GetDatabaseName(), data_table_name); + Result response = api_->GetTable(data_identifier); + if (response.ok()) { + return true; + } + if (response.status().IsNotExist()) { + return false; + } + return response.status(); +} + +Result RestCatalog::GetTableLocation(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + Identifier data_identifier(identifier.GetDatabaseName(), data_table_name); + PAIMON_ASSIGN_OR_RAISE(GetTableResponse response, api_->GetTable(data_identifier)); + return response.GetPath(); +} + +Status RestCatalog::DropTable(const Identifier& identifier, bool ignore_if_not_exists) { + PAIMON_RETURN_NOT_OK(CheckNotSystemTable(identifier, "drop table")); + PAIMON_RETURN_NOT_OK(CheckNotBranch(identifier, "drop table")); + Status status = api_->DropTable(identifier); + if (status.IsNotExist() && ignore_if_not_exists) { + return Status::OK(); + } + return status; +} + +Status RestCatalog::RenameTable(const Identifier& from_table, const Identifier& to_table, + bool ignore_if_not_exists) { + PAIMON_RETURN_NOT_OK(CheckNotSystemTable(from_table, "rename table")); + PAIMON_RETURN_NOT_OK(CheckNotSystemTable(to_table, "rename table")); + PAIMON_RETURN_NOT_OK(CheckNotBranch(from_table, "rename table")); + PAIMON_RETURN_NOT_OK(CheckNotBranch(to_table, "rename table")); + Status status = api_->RenameTable(from_table, to_table); + if (status.IsNotExist() && ignore_if_not_exists) { + return Status::OK(); + } + return status; +} + +Result> RestCatalog::ToTableSchema( + const GetTableResponse& response, const std::optional& branch) { + std::string table_schema_json; + try { + rapidjson::Document src; + src.Parse(response.GetSchemaJson().c_str()); + if (src.HasParseError() || !src.IsObject() || !src.HasMember("fields") || + !src["fields"].IsArray()) { + return Status::Invalid("invalid table schema json from the rest server"); + } + rapidjson::Document out; + out.SetObject(); + rapidjson::Document::AllocatorType& allocator = out.GetAllocator(); + out.AddMember(rapidjson::StringRef("version"), TableSchema::CURRENT_VERSION, allocator); + out.AddMember(rapidjson::StringRef("id"), response.GetSchemaId(), allocator); + rapidjson::Value fields(rapidjson::kArrayType); + fields.CopyFrom(src["fields"], allocator); + out.AddMember(rapidjson::StringRef("highestFieldId"), ComputeHighestFieldId(src["fields"]), + allocator); + out.AddMember(rapidjson::StringRef("fields"), fields.Move(), allocator); + for (const char* key : {"partitionKeys", "primaryKeys"}) { + rapidjson::Value keys(rapidjson::kArrayType); + if (src.HasMember(key) && src[key].IsArray()) { + keys.CopyFrom(src[key], allocator); + } + out.AddMember(rapidjson::StringRef(key), keys.Move(), allocator); + } + std::map options_map; + if (src.HasMember("options") && src["options"].IsObject()) { + options_map = + RapidJsonUtil::DeserializeValue>(src["options"]); + } + options_map[kPathOption] = response.GetPath(); + response.GetAuditFields().PutAuditOptionsTo(&options_map); + if (branch) { + options_map[Options::BRANCH] = branch.value(); + } + out.AddMember(rapidjson::StringRef("options"), + RapidJsonUtil::SerializeValue(options_map, &allocator).Move(), allocator); + if (src.HasMember("comment") && src["comment"].IsString()) { + rapidjson::Value comment; + comment.CopyFrom(src["comment"], allocator); + out.AddMember(rapidjson::StringRef("comment"), comment.Move(), allocator); + } + out.AddMember(rapidjson::StringRef("timeMillis"), + response.GetAuditFields().updated_at.value_or(0), allocator); + table_schema_json = RestUtil::JsonToString(out); + } catch (const std::exception& e) { + return Status::Invalid("failed to convert rest table schema: ", e.what()); + } + return TableSchema::CreateFromJson(table_schema_json); +} + +Result> RestCatalog::LoadDataTableSchema( + const Identifier& data_identifier, const std::optional& branch, + std::string* table_path) const { + PAIMON_ASSIGN_OR_RAISE(GetTableResponse response, api_->GetTable(data_identifier)); + if (table_path != nullptr) { + *table_path = response.GetPath(); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr schema, ToTableSchema(response, branch)); + return std::shared_ptr(std::move(schema)); +} + +Result> RestCatalog::LoadTableSchema(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + PAIMON_ASSIGN_OR_RAISE(std::optional branch, identifier.GetBranchName()); + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + Identifier data_identifier(identifier.GetDatabaseName(), data_table_name); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::optional system_table_name, + identifier.GetSystemTableName()); + if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) { + return Status::NotExist(fmt::format("{} not exist", identifier.ToString())); + } + std::string table_path; + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr latest_schema, + LoadDataTableSchema(data_identifier, branch, &table_path)); + std::map dynamic_options; + if (branch) { + dynamic_options[Options::BRANCH] = branch.value(); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr system_table, + SystemTableLoader::Load(system_table_name.value(), fs_, table_path, + latest_schema, dynamic_options)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr arrow_schema, + system_table->ArrowSchema()); + return std::make_shared(std::move(arrow_schema)); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, + LoadDataTableSchema(data_identifier, branch, nullptr)); + return std::static_pointer_cast(schema); +} + +Result> RestCatalog::GetTable(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, LoadTableSchema(identifier)); + return std::make_shared(schema, identifier.GetDatabaseName(), identifier.GetTableName()); +} + +std::string RestCatalog::GetRootPath() const { + return warehouse_; +} + +std::shared_ptr RestCatalog::GetFileSystem() const { + return fs_; +} + +Result> RestCatalog::ListSnapshots(const Identifier& identifier, + const std::string& branch) const { + if (!branch.empty()) { + return Status::NotImplemented( + "do not support listing snapshots of a branch for the rest catalog."); + } + PAIMON_RETURN_NOT_OK(CheckNotSystemTable(identifier, "list snapshots")); + PAIMON_ASSIGN_OR_RAISE(std::vector snapshots, api_->ListSnapshots(identifier)); + std::sort(snapshots.begin(), snapshots.end(), + [](const Snapshot& a, const Snapshot& b) { return a.Id() < b.Id(); }); + std::vector result; + result.reserve(snapshots.size()); + for (const auto& snapshot : snapshots) { + result.push_back(snapshot.ToSnapshotInfo()); + } + return result; +} + +} // namespace paimon diff --git a/src/paimon/rest/rest_catalog.h b/src/paimon/rest/rest_catalog.h new file mode 100644 index 000000000..9a5c430dd --- /dev/null +++ b/src/paimon/rest/rest_catalog.h @@ -0,0 +1,105 @@ +/* + * 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 +#include +#include +#include +#include + +#include "paimon/catalog/catalog.h" +#include "paimon/logging.h" +#include "paimon/rest/rest_api.h" +#include "paimon/result.h" +#include "paimon/status.h" + +struct ArrowSchema; + +namespace paimon { +class FileSystem; +class TableSchema; + +/// A catalog backed by a REST catalog server, mirroring `RESTCatalog` in the Java +/// implementation. Metadata operations are delegated to `RestApi`; table data is +/// accessed through the file system configured by the (server merged) options. +class RestCatalog : public Catalog { + public: + /// Creates the catalog: fetches and merges "/v1/config" from the server configured + /// by `Options::URI`, then builds the file system from the merged options. + /// + /// @param warehouse The warehouse identifier sent to the server; may be empty. + static Result> Create( + const std::string& warehouse, const std::map& options, + const std::shared_ptr& file_system, + const HttpClient::Config& http_config = HttpClient::Config()); + + Status CreateDatabase(const std::string& name, + const std::map& options, + bool ignore_if_exists) override; + Status CreateTable(const Identifier& identifier, ArrowSchema* c_schema, + const std::vector& partition_keys, + const std::vector& primary_keys, + const std::map& options, + bool ignore_if_exists) override; + Status DropDatabase(const std::string& name, bool ignore_if_not_exists, bool cascade) override; + Status DropTable(const Identifier& identifier, bool ignore_if_not_exists) override; + Status RenameTable(const Identifier& from_table, const Identifier& to_table, + bool ignore_if_not_exists) override; + Result> ListDatabases() const override; + Result> ListTables(const std::string& db_name) const override; + Result DatabaseExists(const std::string& db_name) const override; + Result TableExists(const Identifier& identifier) const override; + std::string GetDatabaseLocation(const std::string& db_name) const override; + Result GetTableLocation(const Identifier& identifier) const override; + Result> LoadTableSchema(const Identifier& identifier) const override; + std::string GetRootPath() const override; + std::shared_ptr GetFileSystem() const override; + Result> GetTable(const Identifier& identifier) const override; + Result> ListSnapshots(const Identifier& identifier, + const std::string& branch) const override; + + /// Options merged with the server side config. + const std::map& GetMergedOptions() const; + + private: + RestCatalog(std::unique_ptr api, const std::shared_ptr& fs, + const std::string& warehouse); + + /// Loads the table from the server and converts the response to a `TableSchema` + /// (options are enriched with the table path, audit info and branch). + Result> LoadDataTableSchema( + const Identifier& data_identifier, const std::optional& branch, + std::string* table_path) const; + + static Result> ToTableSchema( + const GetTableResponse& response, const std::optional& branch); + + static bool IsSystemDatabase(const std::string& db_name); + static Status CheckNotSystemTable(const Identifier& identifier, const std::string& action); + static Status CheckNotBranch(const Identifier& identifier, const std::string& action); + + std::unique_ptr api_; + std::shared_ptr fs_; + std::string warehouse_; + /// The "table-default." options of the merged config, applied to `CreateTable` + /// options when absent. + std::map table_default_options_; + std::shared_ptr logger_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/rest_catalog_test.cpp b/src/paimon/rest/rest_catalog_test.cpp new file mode 100644 index 000000000..144170f13 --- /dev/null +++ b/src/paimon/rest/rest_catalog_test.cpp @@ -0,0 +1,624 @@ +/* + * 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/rest/rest_catalog.h" + +#include +#include +#include +#include + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "gtest/gtest.h" +#include "paimon/catalog/catalog.h" +#include "paimon/catalog/table.h" +#include "paimon/defs.h" +#include "paimon/rest/mock_rest_server.h" +#include "paimon/rest/rest_api.h" +#include "paimon/schema/schema.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +namespace { + +constexpr const char kToken[] = "test-token"; +constexpr const char kPrefix[] = "paimon"; +constexpr const char kWarehouse[] = "wh1"; + +/// The in-memory catalog state behind the mock rest server. +struct MockCatalogState { + struct TableData { + std::string schema_json; + int64_t schema_id = 0; + std::string path; + }; + // database name -> table name -> table + std::map> databases; + // headers of the last request, with lower-cased names + std::map last_headers; +}; + +MockRestServer::Response JsonResponse(int32_t code, const std::string& body) { + MockRestServer::Response response; + response.code = code; + response.body = body; + return response; +} + +MockRestServer::Response MockError(int32_t code, const std::string& resource_type, + const std::string& resource_name, const std::string& message) { + ErrorResponse error(resource_type, resource_name, message, code); + return JsonResponse(code, error.ToJsonString().value()); +} + +std::string SnapshotJson(int64_t id) { + return fmt::format( + R"({{"version":3,"id":{},"schemaId":0,"baseManifestList":"bml","deltaManifestList":"dml",)" + R"("commitUser":"user1","commitIdentifier":1,"commitKind":"APPEND","timeMillis":100,)" + R"("totalRecordCount":10,"deltaRecordCount":1}})", + id); +} + +std::string TableResponseJson(const std::string& name, const MockCatalogState::TableData& table) { + return fmt::format( + R"({{"id":"1","name":"{}","path":"{}","isExternal":false,"schemaId":{},"schema":{},)" + R"("owner":"owner1","updatedAt":123}})", + name, table.path, table.schema_id, table.schema_json); +} + +/// Implements the subset of the rest catalog protocol used by `RestCatalog` on top of +/// `MockCatalogState`. +MockRestServer::Response HandleCatalogRequest(MockCatalogState* state, + const MockRestServer::Request& request) { + state->last_headers = request.headers; + auto auth_iter = request.headers.find("authorization"); + if (auth_iter == request.headers.end() || + auth_iter->second != std::string("Bearer ") + kToken) { + return MockError(401, "", "", "invalid token"); + } + if (request.path == "/v1/config") { + auto warehouse_iter = request.query_params.find("warehouse"); + if (warehouse_iter == request.query_params.end() || warehouse_iter->second != kWarehouse) { + return MockError(400, "", "", "unexpected warehouse"); + } + ConfigResponse config({{RestApi::kOptionPrefix, kPrefix}, + {"header.x-server-header", "from-config"}, + {"table-default.write-only", "true"}, + {"table-default.bucket", "8"}}, + {{"server-override", "from-server"}}); + return JsonResponse(200, config.ToJsonString().value()); + } + const std::string base = std::string("/v1/") + kPrefix; + if (request.path.rfind(base, 0) != 0) { + return MockError(404, "", "", "unknown path " + request.path); + } + std::string rest = request.path.substr(base.size()); + + if (rest == "/databases") { + if (request.method == "GET") { + // one database per page to exercise the pagination loop + std::vector names; + for (const auto& [name, tables] : state->databases) { + names.push_back(name); + } + size_t index = 0; + auto token_iter = request.query_params.find(RestApi::kQueryParamPageToken); + if (token_iter != request.query_params.end()) { + index = std::stoul(token_iter->second); + } + std::vector page; + std::optional next_page_token; + if (index < names.size()) { + page.push_back(names[index]); + if (index + 1 < names.size()) { + next_page_token = std::to_string(index + 1); + } + } + ListDatabasesResponse response(page, next_page_token); + return JsonResponse(200, response.ToJsonString().value()); + } + if (request.method == "POST") { + CreateDatabaseRequest create_request("", {}); + if (!RapidJsonUtil::FromJsonString(request.body, &create_request).ok()) { + return MockError(400, "", "", "bad create database request"); + } + if (state->databases.count(create_request.GetName()) > 0) { + return MockError(409, ErrorResponse::kResourceTypeDatabase, + create_request.GetName(), "database already exists"); + } + state->databases[create_request.GetName()] = {}; + return JsonResponse(200, ""); + } + return MockError(400, "", "", "unsupported method"); + } + + if (rest == "/tables/rename" && request.method == "POST") { + RenameTableRequest rename_request("", "", "", ""); + if (!RapidJsonUtil::FromJsonString(request.body, &rename_request).ok()) { + return MockError(400, "", "", "bad rename table request"); + } + auto db_iter = state->databases.find(rename_request.GetSourceDatabase()); + if (db_iter == state->databases.end() || + db_iter->second.count(rename_request.GetSourceTable()) == 0) { + return MockError(404, ErrorResponse::kResourceTypeTable, + rename_request.GetSourceTable(), "table not found"); + } + auto& dest_tables = state->databases[rename_request.GetDestinationDatabase()]; + if (dest_tables.count(rename_request.GetDestinationTable()) > 0) { + return MockError(409, ErrorResponse::kResourceTypeTable, + rename_request.GetDestinationTable(), "table already exists"); + } + dest_tables[rename_request.GetDestinationTable()] = + db_iter->second[rename_request.GetSourceTable()]; + db_iter->second.erase(rename_request.GetSourceTable()); + return JsonResponse(200, ""); + } + + const std::string databases_prefix = "/databases/"; + if (rest.rfind(databases_prefix, 0) != 0) { + return MockError(404, "", "", "unknown path " + request.path); + } + std::string remainder = rest.substr(databases_prefix.size()); + size_t tables_pos = remainder.find("/tables"); + + if (tables_pos == std::string::npos) { + const std::string& db_name = remainder; + auto db_iter = state->databases.find(db_name); + if (request.method == "GET") { + if (db_iter == state->databases.end()) { + return MockError(404, ErrorResponse::kResourceTypeDatabase, db_name, + "database not found"); + } + std::string body = fmt::format( + R"({{"id":"1","name":"{}","location":"{}/{}.db","options":{{"dbk":"dbv"}}}})", + db_name, kWarehouse, db_name); + return JsonResponse(200, body); + } + if (request.method == "DELETE") { + if (db_iter == state->databases.end()) { + return MockError(404, ErrorResponse::kResourceTypeDatabase, db_name, + "database not found"); + } + state->databases.erase(db_iter); + return JsonResponse(200, ""); + } + return MockError(400, "", "", "unsupported method"); + } + + std::string db_name = remainder.substr(0, tables_pos); + auto db_iter = state->databases.find(db_name); + if (db_iter == state->databases.end()) { + return MockError(404, ErrorResponse::kResourceTypeDatabase, db_name, "database not found"); + } + auto& tables = db_iter->second; + std::string table_part = remainder.substr(tables_pos + std::strlen("/tables")); + + if (table_part.empty()) { + if (request.method == "GET") { + std::vector names; + for (const auto& [name, table] : tables) { + names.push_back(name); + } + ListTablesResponse response(names, std::nullopt); + return JsonResponse(200, response.ToJsonString().value()); + } + if (request.method == "POST") { + CreateTableRequest create_request("", "", ""); + if (!RapidJsonUtil::FromJsonString(request.body, &create_request).ok()) { + return MockError(400, "", "", "bad create table request"); + } + if (tables.count(create_request.GetTable()) > 0) { + return MockError(409, ErrorResponse::kResourceTypeTable, create_request.GetTable(), + "table already exists"); + } + MockCatalogState::TableData table; + table.schema_json = create_request.GetSchemaJson(); + table.schema_id = 0; + table.path = fmt::format("{}/{}.db/{}", kWarehouse, db_name, create_request.GetTable()); + tables[create_request.GetTable()] = table; + return JsonResponse(200, ""); + } + return MockError(400, "", "", "unsupported method"); + } + + // "/
" or "/
/snapshots" + std::string table_name = table_part.substr(1); + bool list_snapshots = false; + const std::string snapshots_suffix = "/snapshots"; + if (table_name.size() > snapshots_suffix.size() && + table_name.compare(table_name.size() - snapshots_suffix.size(), snapshots_suffix.size(), + snapshots_suffix) == 0) { + table_name = table_name.substr(0, table_name.size() - snapshots_suffix.size()); + list_snapshots = true; + } + auto table_iter = tables.find(table_name); + if (table_iter == tables.end()) { + return MockError(404, ErrorResponse::kResourceTypeTable, table_name, "table not found"); + } + if (list_snapshots) { + // two pages, out of order to exercise pagination and sorting + auto token_iter = request.query_params.find(RestApi::kQueryParamPageToken); + if (token_iter == request.query_params.end()) { + return JsonResponse( + 200, fmt::format(R"({{"snapshots":[{}],"nextPageToken":"1"}})", SnapshotJson(2))); + } + return JsonResponse(200, fmt::format(R"({{"snapshots":[{}]}})", SnapshotJson(1))); + } + if (request.method == "GET") { + return JsonResponse(200, TableResponseJson(table_name, table_iter->second)); + } + if (request.method == "DELETE") { + tables.erase(table_iter); + return JsonResponse(200, ""); + } + return MockError(400, "", "", "unsupported method"); +} + +} // namespace + +class RestCatalogTest : public ::testing::Test { + protected: + void SetUp() override { + state_ = std::make_shared(); + ASSERT_OK_AND_ASSIGN( + server_, MockRestServer::Start([state = state_](const MockRestServer::Request& req) { + return HandleCatalogRequest(state.get(), req); + })); + options_ = { + {Options::METASTORE, "rest"}, + {Options::URI, server_->GetBaseUri()}, + {Options::TOKEN_PROVIDER, "bear"}, + {Options::TOKEN, kToken}, + {Options::FILE_SYSTEM, "local"}, + // mock_format is linked statically into the test binary, so its factory is + // registered in the binary's own registry even when the real format plugin + // dylibs register into a different one (macOS two-level namespace) + {Options::FILE_FORMAT, "mock_format"}, + {Options::MANIFEST_FORMAT, "mock_format"}, + {"header.x-client-header", "from-client"}, + }; + } + + void TearDown() override { + if (server_) { + server_->Stop(); + } + } + + Result> CreateRestCatalog() { + return RestCatalog::Create(kWarehouse, options_, nullptr); + } + + Status CreateSampleTable(Catalog* catalog, const Identifier& identifier) { + std::shared_ptr schema = + arrow::schema({arrow::field("f0", arrow::int32(), /*nullable=*/false), + arrow::field("f1", arrow::utf8())}); + struct ArrowSchema c_schema; + if (!arrow::ExportSchema(*schema, &c_schema).ok()) { + return Status::Invalid("failed to export arrow schema"); + } + Status status = catalog->CreateTable(identifier, &c_schema, /*partition_keys=*/{}, + /*primary_keys=*/{"f0"}, {{"bucket", "2"}}, + /*ignore_if_exists=*/false); + // CreateTable takes ownership of the exported schema only once it reaches + // arrow::ImportSchema, which an identifier rejected by its checks never does + if (c_schema.release != nullptr) { + c_schema.release(&c_schema); + } + return status; + } + + std::shared_ptr state_; + std::unique_ptr server_; + std::map options_; +}; + +TEST_F(RestCatalogTest, CreateMergesServerConfig) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + const std::map& merged = catalog->GetMergedOptions(); + ASSERT_EQ(kPrefix, merged.at(RestApi::kOptionPrefix)); + ASSERT_EQ("from-server", merged.at("server-override")); + ASSERT_EQ(kWarehouse, catalog->GetRootPath()); + ASSERT_NE(nullptr, catalog->GetFileSystem()); +} + +TEST_F(RestCatalogTest, CreateViaCatalogFactory) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, Catalog::Create(kWarehouse, options_)); + ASSERT_OK_AND_ASSIGN(std::vector databases, catalog->ListDatabases()); + ASSERT_TRUE(databases.empty()); +} + +TEST_F(RestCatalogTest, CreateWithWrongTokenFails) { + options_[Options::TOKEN] = "wrong-token"; + ASSERT_NOK_WITH_MSG(CreateRestCatalog().status(), "not authorized"); +} + +TEST_F(RestCatalogTest, CreateRejectsInvalidOptions) { + // all rejected by client side validation, before any request reaches the server + const std::map valid_options = options_; + + options_.erase(Options::URI); + ASSERT_NOK_WITH_MSG(CreateRestCatalog().status(), "'uri' must be configured"); + + options_ = valid_options; + options_.erase(Options::TOKEN_PROVIDER); + ASSERT_NOK_WITH_MSG(CreateRestCatalog().status(), "'token.provider' must be configured"); + + options_ = valid_options; + options_[Options::TOKEN_PROVIDER] = "dlf"; + Status unsupported_provider = CreateRestCatalog().status(); + ASSERT_TRUE(unsupported_provider.IsNotImplemented()) << unsupported_provider.ToString(); + ASSERT_NOK_WITH_MSG(unsupported_provider, "unsupported token provider"); + + options_ = valid_options; + options_.erase(Options::TOKEN); + ASSERT_NOK_WITH_MSG(CreateRestCatalog().status(), "bear token provider"); +} + +TEST_F(RestCatalogTest, DatabaseOperations) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + + ASSERT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/false)); + ASSERT_OK(catalog->CreateDatabase("db2", {}, /*ignore_if_exists=*/false)); + Status duplicated = catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/false); + ASSERT_TRUE(duplicated.IsExist()) << duplicated.ToString(); + ASSERT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/true)); + + // the mock server returns one database per page + ASSERT_OK_AND_ASSIGN(std::vector databases, catalog->ListDatabases()); + ASSERT_EQ((std::vector{"db1", "db2"}), databases); + + ASSERT_OK_AND_ASSIGN(bool exists, catalog->DatabaseExists("db1")); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog->DatabaseExists("db3")); + ASSERT_FALSE(exists); + + ASSERT_EQ("wh1/db1.db", catalog->GetDatabaseLocation("db1")); + ASSERT_EQ("", catalog->GetDatabaseLocation("db3")); + + ASSERT_OK(catalog->DropDatabase("db2", /*ignore_if_not_exists=*/false, /*cascade=*/false)); + ASSERT_OK(catalog->DropDatabase("db2", /*ignore_if_not_exists=*/true, /*cascade=*/false)); + Status missing = catalog->DropDatabase("db2", /*ignore_if_not_exists=*/false, + /*cascade=*/false); + ASSERT_TRUE(missing.IsNotExist()) << missing.ToString(); + + ASSERT_OK(CreateSampleTable(catalog.get(), Identifier("db1", "t1"))); + ASSERT_NOK_WITH_MSG( + catalog->DropDatabase("db1", /*ignore_if_not_exists=*/false, /*cascade=*/false), + "non-empty database"); + // cascade drop skips the emptiness check + ASSERT_OK(catalog->DropDatabase("db1", /*ignore_if_not_exists=*/false, /*cascade=*/true)); + ASSERT_OK_AND_ASSIGN(exists, catalog->DatabaseExists("db1")); + ASSERT_FALSE(exists); +} + +TEST_F(RestCatalogTest, TableOperations) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/false)); + Identifier identifier("db1", "t1"); + + Status missing_db = CreateSampleTable(catalog.get(), Identifier("db_missing", "t1")); + ASSERT_TRUE(missing_db.IsNotExist()) << missing_db.ToString(); + + ASSERT_OK(CreateSampleTable(catalog.get(), identifier)); + Status duplicated = CreateSampleTable(catalog.get(), identifier); + ASSERT_TRUE(duplicated.IsExist()) << duplicated.ToString(); + + ASSERT_OK_AND_ASSIGN(std::vector tables, catalog->ListTables("db1")); + ASSERT_EQ((std::vector{"t1"}), tables); + Status list_missing = catalog->ListTables("db_missing").status(); + ASSERT_TRUE(list_missing.IsNotExist()) << list_missing.ToString(); + + ASSERT_OK_AND_ASSIGN(bool exists, catalog->TableExists(identifier)); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog->TableExists(Identifier("db1", "t2"))); + ASSERT_FALSE(exists); + + ASSERT_OK_AND_ASSIGN(std::string location, catalog->GetTableLocation(identifier)); + ASSERT_EQ("wh1/db1.db/t1", location); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr
table, catalog->GetTable(identifier)); + ASSERT_EQ("t1", table->Name()); + std::shared_ptr schema = + std::dynamic_pointer_cast(table->LatestSchema()); + ASSERT_NE(nullptr, schema); + ASSERT_EQ((std::vector{"f0", "f1"}), schema->FieldNames()); + ASSERT_EQ((std::vector{"f0"}), schema->PrimaryKeys()); + ASSERT_EQ(0, schema->Id()); + // options are enriched with the table path and audit info from the server + ASSERT_EQ("wh1/db1.db/t1", schema->Options().at("path")); + ASSERT_EQ("owner1", schema->Options().at("owner")); + + // "table-default." options of the merged config apply only where the caller left the + // option unset: "write-only" is taken from the config, "bucket" keeps the value passed + // to CreateTable instead of the configured "table-default.bucket" of 8 + ASSERT_EQ("true", schema->Options().at("write-only")); + ASSERT_EQ("2", schema->Options().at("bucket")); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr loaded_schema, + catalog->LoadTableSchema(identifier)); + ASSERT_EQ((std::vector{"f0", "f1"}), loaded_schema->FieldNames()); + Status schema_missing = catalog->LoadTableSchema(Identifier("db1", "t2")).status(); + ASSERT_TRUE(schema_missing.IsNotExist()) << schema_missing.ToString(); + + ASSERT_OK(catalog->RenameTable(identifier, Identifier("db1", "t2"), + /*ignore_if_not_exists=*/false)); + ASSERT_OK_AND_ASSIGN(exists, catalog->TableExists(Identifier("db1", "t2"))); + ASSERT_TRUE(exists); + ASSERT_OK(catalog->RenameTable(identifier, Identifier("db1", "t3"), + /*ignore_if_not_exists=*/true)); + Status rename_missing = catalog->RenameTable(identifier, Identifier("db1", "t3"), + /*ignore_if_not_exists=*/false); + ASSERT_TRUE(rename_missing.IsNotExist()) << rename_missing.ToString(); + + ASSERT_OK(catalog->DropTable(Identifier("db1", "t2"), /*ignore_if_not_exists=*/false)); + ASSERT_OK(catalog->DropTable(Identifier("db1", "t2"), /*ignore_if_not_exists=*/true)); + Status drop_missing = catalog->DropTable(Identifier("db1", "t2"), + /*ignore_if_not_exists=*/false); + ASSERT_TRUE(drop_missing.IsNotExist()) << drop_missing.ToString(); +} + +TEST_F(RestCatalogTest, ClientAndServerHeadersAreSent) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_OK_AND_ASSIGN(std::vector databases, catalog->ListDatabases()); + // "header." options from both the client and the merged server config are sent as + // http headers on every request + ASSERT_EQ("from-client", state_->last_headers.at("x-client-header")); + ASSERT_EQ("from-config", state_->last_headers.at("x-server-header")); + ASSERT_EQ(std::string("Bearer ") + kToken, state_->last_headers.at("authorization")); +} + +TEST_F(RestCatalogTest, SystemTableSchema) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/false)); + ASSERT_OK(CreateSampleTable(catalog.get(), Identifier("db1", "t1"))); + + // the "options" system table has a static schema and needs no file system access + Identifier system_identifier("db1", "t1$options"); + ASSERT_OK_AND_ASSIGN(bool exists, catalog->TableExists(system_identifier)); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog->TableExists(Identifier("db1", "t9$options"))); + ASSERT_FALSE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog->TableExists(Identifier("db1", "t1$unsupported"))); + ASSERT_FALSE(exists); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr schema, + catalog->LoadTableSchema(system_identifier)); + ASSERT_EQ((std::vector{"key", "value"}), schema->FieldNames()); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr
table, catalog->GetTable(system_identifier)); + ASSERT_EQ("t1$options", table->Name()); + ASSERT_EQ((std::vector{"key", "value"}), table->LatestSchema()->FieldNames()); + + Status unsupported = catalog->LoadTableSchema(Identifier("db1", "t1$unsupported")).status(); + ASSERT_TRUE(unsupported.IsNotExist()) << unsupported.ToString(); +} + +TEST_F(RestCatalogTest, BranchTableSchemaCarriesBranchOption) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/false)); + ASSERT_OK(CreateSampleTable(catalog.get(), Identifier("db1", "t1"))); + + // "t1$branch_b1" is a branch table, not a system table: the schema of the data + // table is loaded and the branch is injected into the schema options + Identifier branch_identifier("db1", "t1$branch_b1"); + ASSERT_OK_AND_ASSIGN(std::shared_ptr
table, catalog->GetTable(branch_identifier)); + ASSERT_EQ("t1$branch_b1", table->Name()); + std::shared_ptr schema = + std::dynamic_pointer_cast(table->LatestSchema()); + ASSERT_NE(nullptr, schema); + ASSERT_EQ("b1", schema->Options().at(Options::BRANCH)); + + // writing operations reject branch tables, mirroring checkNotBranch in Java + ASSERT_NOK_WITH_MSG(catalog->DropTable(branch_identifier, /*ignore_if_not_exists=*/false), + "branch table"); + ASSERT_NOK_WITH_MSG(catalog->RenameTable(branch_identifier, Identifier("db1", "t2"), + /*ignore_if_not_exists=*/false), + "branch table"); + ASSERT_NOK_WITH_MSG(CreateSampleTable(catalog.get(), Identifier("db1", "t2$branch_b1")), + "branch table"); +} + +TEST_F(RestCatalogTest, NestedSchemaHighestFieldId) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/false)); + // inject a table with nested types directly; ids inside nested rows must be + // included when computing highestFieldId (mirrors RowType.currentHighestFieldId) + MockCatalogState::TableData table_data; + table_data.schema_json = R"({ + "fields": [ + {"id": 0, "name": "f0", "type": "INT NOT NULL"}, + {"id": 1, "name": "s", "type": {"type": "ROW", + "fields": [{"id": 3, "name": "inner", "type": "INT"}]}}, + {"id": 2, "name": "arr", "type": {"type": "ARRAY", + "element": {"type": "ROW", + "fields": [{"id": 7, "name": "deep", "type": "BIGINT"}]}}} + ], + "partitionKeys": [], + "primaryKeys": [], + "options": {} + })"; + table_data.schema_id = 5; + table_data.path = "wh1/db1.db/nested"; + state_->databases["db1"]["nested"] = table_data; + + ASSERT_OK_AND_ASSIGN(std::shared_ptr
table, + catalog->GetTable(Identifier("db1", "nested"))); + std::shared_ptr schema = + std::dynamic_pointer_cast(table->LatestSchema()); + ASSERT_NE(nullptr, schema); + ASSERT_EQ(5, schema->Id()); + ASSERT_EQ(7, schema->HighestFieldId()); + ASSERT_EQ((std::vector{"f0", "s", "arr"}), schema->FieldNames()); +} + +TEST_F(RestCatalogTest, SystemTableChecks) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_NOK_WITH_MSG(catalog->CreateDatabase("sys", {}, false), "system database"); + ASSERT_NOK_WITH_MSG(catalog->DropDatabase("sys", false, false), "system database"); + ASSERT_NOK_WITH_MSG(catalog->DropTable(Identifier("sys", "t"), false), "system table"); + ASSERT_NOK_WITH_MSG( + catalog->RenameTable(Identifier("db1", "t1$snapshots"), Identifier("db1", "t2"), false), + "system table"); +} + +TEST_F(RestCatalogTest, ListSnapshots) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/false)); + Identifier identifier("db1", "t1"); + ASSERT_OK(CreateSampleTable(catalog.get(), identifier)); + + ASSERT_OK_AND_ASSIGN(std::vector snapshots, + catalog->ListSnapshots(identifier, "")); + ASSERT_EQ(2, snapshots.size()); + // fetched via two pages and sorted by snapshot id + ASSERT_EQ(1, snapshots[0].snapshot_id); + ASSERT_EQ(2, snapshots[1].snapshot_id); + ASSERT_EQ("user1", snapshots[0].commit_user); + ASSERT_EQ(SnapshotInfo::CommitKind::APPEND, snapshots[0].commit_kind); + + Status missing = catalog->ListSnapshots(Identifier("db1", "t9"), "").status(); + ASSERT_TRUE(missing.IsNotExist()) << missing.ToString(); + ASSERT_NOK(catalog->ListSnapshots(identifier, "branch1").status()); +} + +TEST(RestApiErrorTest, ErrorToStatus) { + HttpClient::Response response; + response.code = 404; + response.body = R"({"message": "no table", "resourceType": "TABLE", "resourceName": "t1"})"; + response.headers["x-request-id"] = "req-123"; + Status status = RestApi::ErrorToStatus(response); + ASSERT_TRUE(status.IsNotExist()); + ASSERT_TRUE(status.ToString().find("requestId:req-123") != std::string::npos) + << status.ToString(); + + response.code = 409; + Status exist_status = RestApi::ErrorToStatus(response); + ASSERT_TRUE(exist_status.IsExist()); + + response.code = 501; + response.body = ""; + ASSERT_TRUE(RestApi::ErrorToStatus(response).IsNotImplemented()); + + response.code = 500; + response.body = "not-a-json"; + ASSERT_NOK_WITH_MSG(RestApi::ErrorToStatus(response), "server error"); +} + +} // namespace paimon::test diff --git a/src/paimon/rest/rest_messages.cpp b/src/paimon/rest/rest_messages.cpp new file mode 100644 index 000000000..09f77a03d --- /dev/null +++ b/src/paimon/rest/rest_messages.cpp @@ -0,0 +1,373 @@ +/* + * 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/rest/rest_messages.h" + +#include + +#include "paimon/rest/rest_util.h" + +namespace paimon { + +namespace { + +constexpr const char kFieldMessage[] = "message"; +constexpr const char kFieldResourceType[] = "resourceType"; +constexpr const char kFieldResourceName[] = "resourceName"; +constexpr const char kFieldCode[] = "code"; +constexpr const char kFieldDefaults[] = "defaults"; +constexpr const char kFieldOverrides[] = "overrides"; +constexpr const char kFieldOwner[] = "owner"; +constexpr const char kFieldCreatedAt[] = "createdAt"; +constexpr const char kFieldCreatedBy[] = "createdBy"; +constexpr const char kFieldUpdatedAt[] = "updatedAt"; +constexpr const char kFieldUpdatedBy[] = "updatedBy"; +constexpr const char kFieldName[] = "name"; +constexpr const char kFieldOptions[] = "options"; +constexpr const char kFieldId[] = "id"; +constexpr const char kFieldLocation[] = "location"; +constexpr const char kFieldDatabases[] = "databases"; +constexpr const char kFieldTables[] = "tables"; +constexpr const char kFieldSnapshots[] = "snapshots"; +constexpr const char kFieldNextPageToken[] = "nextPageToken"; +constexpr const char kFieldPath[] = "path"; +constexpr const char kFieldIsExternal[] = "isExternal"; +constexpr const char kFieldSchemaId[] = "schemaId"; +constexpr const char kFieldSchema[] = "schema"; +constexpr const char kFieldIdentifier[] = "identifier"; +constexpr const char kFieldDatabase[] = "database"; +constexpr const char kFieldObject[] = "object"; +constexpr const char kFieldSource[] = "source"; +constexpr const char kFieldDestination[] = "destination"; + +void AddOptionalStringMember(rapidjson::Value* obj, const char* key, + const std::optional& value, + rapidjson::Document::AllocatorType* allocator) { + if (value) { + obj->AddMember(rapidjson::StringRef(key), + RapidJsonUtil::SerializeValue(value.value(), allocator).Move(), *allocator); + } +} + +rapidjson::Value SerializeIdentifier(const std::string& database, const std::string& table, + rapidjson::Document::AllocatorType* allocator) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldDatabase), + RapidJsonUtil::SerializeValue(database, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldObject), + RapidJsonUtil::SerializeValue(table, allocator).Move(), *allocator); + return obj; +} + +void DeserializeIdentifier(const rapidjson::Value& obj, const char* key, std::string* database, + std::string* table) { + if (!obj.IsObject() || !obj.HasMember(key) || !obj[key].IsObject()) { + throw std::invalid_argument(std::string("member '") + key + + "' must exist and be an object"); + } + const rapidjson::Value& identifier = obj[key]; + *database = RapidJsonUtil::DeserializeKeyValue(identifier, kFieldDatabase); + *table = RapidJsonUtil::DeserializeKeyValue(identifier, kFieldObject); +} + +std::string DeserializeRawJsonMember(const rapidjson::Value& obj, const char* key) { + if (!obj.IsObject() || !obj.HasMember(key) || !obj[key].IsObject()) { + throw std::invalid_argument(std::string("member '") + key + + "' must exist and be an object"); + } + return RestUtil::JsonToString(obj[key]); +} + +} // namespace + +rapidjson::Value ErrorResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldMessage), + RapidJsonUtil::SerializeValue(message_, allocator).Move(), *allocator); + if (!resource_type_.empty()) { + obj.AddMember(rapidjson::StringRef(kFieldResourceType), + RapidJsonUtil::SerializeValue(resource_type_, allocator).Move(), *allocator); + } + if (!resource_name_.empty()) { + obj.AddMember(rapidjson::StringRef(kFieldResourceName), + RapidJsonUtil::SerializeValue(resource_name_, allocator).Move(), *allocator); + } + obj.AddMember(rapidjson::StringRef(kFieldCode), + RapidJsonUtil::SerializeValue(code_, allocator).Move(), *allocator); + return obj; +} + +void ErrorResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + message_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldMessage, std::string()); + resource_type_ = + RapidJsonUtil::DeserializeKeyValue(obj, kFieldResourceType, std::string()); + resource_name_ = + RapidJsonUtil::DeserializeKeyValue(obj, kFieldResourceName, std::string()); + code_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldCode, 0); +} + +rapidjson::Value ConfigResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldDefaults), + RapidJsonUtil::SerializeValue(defaults_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldOverrides), + RapidJsonUtil::SerializeValue(overrides_, allocator).Move(), *allocator); + return obj; +} + +namespace { +/// The Java server may send null values in `defaults`/`overrides`; `merge` filters them +/// out, so they are skipped here instead of failing the whole parse. +std::map DeserializeStringMapSkippingNulls(const rapidjson::Value& obj, + const char* key) { + std::map result; + if (!obj.IsObject() || !obj.HasMember(key) || obj[key].IsNull()) { + return result; + } + const rapidjson::Value& map_value = obj[key]; + if (!map_value.IsObject()) { + throw std::invalid_argument(std::string("member '") + key + "' must be an object"); + } + for (auto iter = map_value.MemberBegin(); iter != map_value.MemberEnd(); ++iter) { + if (iter->value.IsNull()) { + continue; + } + if (!iter->value.IsString()) { + throw std::invalid_argument(std::string("member '") + key + + "' must only contain string values"); + } + result[iter->name.GetString()] = iter->value.GetString(); + } + return result; +} +} // namespace + +void ConfigResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + defaults_ = DeserializeStringMapSkippingNulls(obj, kFieldDefaults); + overrides_ = DeserializeStringMapSkippingNulls(obj, kFieldOverrides); +} + +std::map ConfigResponse::Merge( + const std::map& client_options) const { + std::map merged = defaults_; + for (const auto& [key, value] : client_options) { + merged[key] = value; + } + for (const auto& [key, value] : overrides_) { + merged[key] = value; + } + return merged; +} + +void RestAuditFields::ParseFrom(const rapidjson::Value& obj) { + owner = RapidJsonUtil::DeserializeKeyValue>(obj, kFieldOwner, + std::nullopt); + created_at = RapidJsonUtil::DeserializeKeyValue>(obj, kFieldCreatedAt, + std::nullopt); + created_by = RapidJsonUtil::DeserializeKeyValue>( + obj, kFieldCreatedBy, std::nullopt); + updated_at = RapidJsonUtil::DeserializeKeyValue>(obj, kFieldUpdatedAt, + std::nullopt); + updated_by = RapidJsonUtil::DeserializeKeyValue>( + obj, kFieldUpdatedBy, std::nullopt); +} + +void RestAuditFields::AddTo(rapidjson::Value* obj, + rapidjson::Document::AllocatorType* allocator) const { + AddOptionalStringMember(obj, kFieldOwner, owner, allocator); + if (created_at) { + obj->AddMember(rapidjson::StringRef(kFieldCreatedAt), + RapidJsonUtil::SerializeValue(created_at.value(), allocator).Move(), + *allocator); + } + AddOptionalStringMember(obj, kFieldCreatedBy, created_by, allocator); + if (updated_at) { + obj->AddMember(rapidjson::StringRef(kFieldUpdatedAt), + RapidJsonUtil::SerializeValue(updated_at.value(), allocator).Move(), + *allocator); + } + AddOptionalStringMember(obj, kFieldUpdatedBy, updated_by, allocator); +} + +void RestAuditFields::PutAuditOptionsTo(std::map* options) const { + if (owner) { + (*options)[kFieldOwner] = owner.value(); + } + if (created_at) { + (*options)[kFieldCreatedAt] = std::to_string(created_at.value()); + } + if (created_by) { + (*options)[kFieldCreatedBy] = created_by.value(); + } + if (updated_at) { + (*options)[kFieldUpdatedAt] = std::to_string(updated_at.value()); + } + if (updated_by) { + (*options)[kFieldUpdatedBy] = updated_by.value(); + } +} + +rapidjson::Value CreateDatabaseRequest::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldName), + RapidJsonUtil::SerializeValue(name_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldOptions), + RapidJsonUtil::SerializeValue(options_, allocator).Move(), *allocator); + return obj; +} + +void CreateDatabaseRequest::FromJson(const rapidjson::Value& obj) noexcept(false) { + name_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldName); + options_ = RapidJsonUtil::DeserializeKeyValue>( + obj, kFieldOptions, {}); +} + +rapidjson::Value GetDatabaseResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldId), + RapidJsonUtil::SerializeValue(id_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldName), + RapidJsonUtil::SerializeValue(name_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldLocation), + RapidJsonUtil::SerializeValue(location_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldOptions), + RapidJsonUtil::SerializeValue(options_, allocator).Move(), *allocator); + audit_.AddTo(&obj, allocator); + return obj; +} + +void GetDatabaseResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + id_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldId, std::string()); + name_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldName); + location_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldLocation, std::string()); + options_ = RapidJsonUtil::DeserializeKeyValue>( + obj, kFieldOptions, {}); + audit_.ParseFrom(obj); +} + +rapidjson::Value ListDatabasesResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldDatabases), + RapidJsonUtil::SerializeValue(databases_, allocator).Move(), *allocator); + AddOptionalStringMember(&obj, kFieldNextPageToken, next_page_token_, allocator); + return obj; +} + +void ListDatabasesResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + databases_ = + RapidJsonUtil::DeserializeKeyValue>(obj, kFieldDatabases, {}); + next_page_token_ = RapidJsonUtil::DeserializeKeyValue>( + obj, kFieldNextPageToken, std::nullopt); +} + +rapidjson::Value ListTablesResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldTables), + RapidJsonUtil::SerializeValue(tables_, allocator).Move(), *allocator); + AddOptionalStringMember(&obj, kFieldNextPageToken, next_page_token_, allocator); + return obj; +} + +void ListTablesResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + tables_ = RapidJsonUtil::DeserializeKeyValue>(obj, kFieldTables, {}); + next_page_token_ = RapidJsonUtil::DeserializeKeyValue>( + obj, kFieldNextPageToken, std::nullopt); +} + +rapidjson::Value ListSnapshotsResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldSnapshots), + RapidJsonUtil::SerializeValue(snapshots_, allocator).Move(), *allocator); + AddOptionalStringMember(&obj, kFieldNextPageToken, next_page_token_, allocator); + return obj; +} + +void ListSnapshotsResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + snapshots_ = + RapidJsonUtil::DeserializeKeyValue>(obj, kFieldSnapshots, {}); + next_page_token_ = RapidJsonUtil::DeserializeKeyValue>( + obj, kFieldNextPageToken, std::nullopt); +} + +rapidjson::Value GetTableResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldId), + RapidJsonUtil::SerializeValue(id_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldName), + RapidJsonUtil::SerializeValue(name_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldPath), + RapidJsonUtil::SerializeValue(path_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldIsExternal), + RapidJsonUtil::SerializeValue(is_external_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldSchemaId), + RapidJsonUtil::SerializeValue(schema_id_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldSchema), + RestUtil::ParseToValue(schema_json_, allocator).Move(), *allocator); + audit_.AddTo(&obj, allocator); + return obj; +} + +void GetTableResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + id_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldId, std::string()); + name_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldName, std::string()); + path_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldPath); + is_external_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldIsExternal, false); + schema_id_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldSchemaId); + schema_json_ = DeserializeRawJsonMember(obj, kFieldSchema); + audit_.ParseFrom(obj); +} + +rapidjson::Value CreateTableRequest::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldIdentifier), + SerializeIdentifier(database_, table_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldSchema), + RestUtil::ParseToValue(schema_json_, allocator).Move(), *allocator); + return obj; +} + +void CreateTableRequest::FromJson(const rapidjson::Value& obj) noexcept(false) { + DeserializeIdentifier(obj, kFieldIdentifier, &database_, &table_); + schema_json_ = DeserializeRawJsonMember(obj, kFieldSchema); +} + +rapidjson::Value RenameTableRequest::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldSource), + SerializeIdentifier(source_database_, source_table_, allocator).Move(), + *allocator); + obj.AddMember(rapidjson::StringRef(kFieldDestination), + SerializeIdentifier(destination_database_, destination_table_, allocator).Move(), + *allocator); + return obj; +} + +void RenameTableRequest::FromJson(const rapidjson::Value& obj) noexcept(false) { + DeserializeIdentifier(obj, kFieldSource, &source_database_, &source_table_); + DeserializeIdentifier(obj, kFieldDestination, &destination_database_, &destination_table_); +} + +} // namespace paimon diff --git a/src/paimon/rest/rest_messages.h b/src/paimon/rest/rest_messages.h new file mode 100644 index 000000000..d868d1e5c --- /dev/null +++ b/src/paimon/rest/rest_messages.h @@ -0,0 +1,377 @@ +/* + * 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 +#include +#include +#include +#include + +#include "paimon/common/utils/jsonizable.h" +#include "paimon/common/utils/rapidjson_util.h" +#include "paimon/core/snapshot.h" +#include "rapidjson/allocators.h" +#include "rapidjson/document.h" +#include "rapidjson/rapidjson.h" + +namespace paimon { + +/// Request and response objects of the REST catalog protocol. Each class corresponds to +/// the same-named class under `org.apache.paimon.rest.requests` or +/// `org.apache.paimon.rest.responses` in the Java implementation and the JSON field +/// names must stay aligned. + +class ErrorResponse : public Jsonizable { + public: + static constexpr const char* kResourceTypeDatabase = "DATABASE"; + static constexpr const char* kResourceTypeTable = "TABLE"; + + ErrorResponse(const std::string& resource_type, const std::string& resource_name, + const std::string& message, int32_t code) + : resource_type_(resource_type), + resource_name_(resource_name), + message_(message), + code_(code) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::string& GetResourceType() const { + return resource_type_; + } + const std::string& GetResourceName() const { + return resource_name_; + } + const std::string& GetMessage() const { + return message_; + } + int32_t GetCode() const { + return code_; + } + + ErrorResponse() = default; + + private: + std::string resource_type_; + std::string resource_name_; + std::string message_; + int32_t code_ = 0; +}; + +/// The response of "/v1/config". +class ConfigResponse : public Jsonizable { + public: + ConfigResponse(const std::map& defaults, + const std::map& overrides) + : defaults_(defaults), overrides_(overrides) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + /// Merges with the client options; the precedence is + /// `overrides` > `client_options` > `defaults`. + std::map Merge( + const std::map& client_options) const; + + const std::map& GetDefaults() const { + return defaults_; + } + const std::map& GetOverrides() const { + return overrides_; + } + + ConfigResponse() = default; + + private: + std::map defaults_; + std::map overrides_; +}; + +/// The audit fields shared by database/table responses, mirroring `AuditRESTResponse`. +struct RestAuditFields { + std::optional owner; + std::optional created_at; + std::optional created_by; + std::optional updated_at; + std::optional updated_by; + + void ParseFrom(const rapidjson::Value& obj); + void AddTo(rapidjson::Value* obj, rapidjson::Document::AllocatorType* allocator) const; + /// Adds the present fields to `options`, mirroring `putAuditOptionsTo`. + void PutAuditOptionsTo(std::map* options) const; +}; + +class CreateDatabaseRequest : public Jsonizable { + public: + CreateDatabaseRequest(const std::string& name, + const std::map& options) + : name_(name), options_(options) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::string& GetName() const { + return name_; + } + const std::map& GetOptions() const { + return options_; + } + + CreateDatabaseRequest() = default; + + private: + std::string name_; + std::map options_; +}; + +class GetDatabaseResponse : public Jsonizable { + public: + GetDatabaseResponse(const std::string& id, const std::string& name, const std::string& location, + const std::map& options) + : id_(id), name_(name), location_(location), options_(options) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::string& GetId() const { + return id_; + } + const std::string& GetName() const { + return name_; + } + const std::string& GetLocation() const { + return location_; + } + const std::map& GetOptions() const { + return options_; + } + const RestAuditFields& GetAuditFields() const { + return audit_; + } + + GetDatabaseResponse() = default; + + private: + std::string id_; + std::string name_; + std::string location_; + std::map options_; + RestAuditFields audit_; +}; + +class ListDatabasesResponse : public Jsonizable { + public: + using ItemType = std::string; + + ListDatabasesResponse(const std::vector& databases, + const std::optional& next_page_token) + : databases_(databases), next_page_token_(next_page_token) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::vector& Data() const { + return databases_; + } + const std::optional& NextPageToken() const { + return next_page_token_; + } + + ListDatabasesResponse() = default; + + private: + std::vector databases_; + std::optional next_page_token_; +}; + +class ListTablesResponse : public Jsonizable { + public: + using ItemType = std::string; + + ListTablesResponse(const std::vector& tables, + const std::optional& next_page_token) + : tables_(tables), next_page_token_(next_page_token) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::vector& Data() const { + return tables_; + } + const std::optional& NextPageToken() const { + return next_page_token_; + } + + ListTablesResponse() = default; + + private: + std::vector tables_; + std::optional next_page_token_; +}; + +/// Each element is a snapshot in the same JSON layout as the snapshot files. +class ListSnapshotsResponse : public Jsonizable { + public: + using ItemType = Snapshot; + + ListSnapshotsResponse(const std::vector& snapshots, + const std::optional& next_page_token) + : snapshots_(snapshots), next_page_token_(next_page_token) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::vector& Data() const { + return snapshots_; + } + const std::optional& NextPageToken() const { + return next_page_token_; + } + + ListSnapshotsResponse() = default; + + private: + std::vector snapshots_; + std::optional next_page_token_; +}; + +/// The nested `schema` object (the Java `org.apache.paimon.schema.Schema` layout: +/// fields/partitionKeys/primaryKeys/options/comment) is kept as a raw JSON string and +/// converted to a `TableSchema` by the catalog. +class GetTableResponse : public Jsonizable { + public: + GetTableResponse(const std::string& id, const std::string& name, const std::string& path, + bool is_external, int64_t schema_id, const std::string& schema_json) + : id_(id), + name_(name), + path_(path), + is_external_(is_external), + schema_id_(schema_id), + schema_json_(schema_json) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::string& GetId() const { + return id_; + } + const std::string& GetName() const { + return name_; + } + const std::string& GetPath() const { + return path_; + } + bool IsExternal() const { + return is_external_; + } + int64_t GetSchemaId() const { + return schema_id_; + } + const std::string& GetSchemaJson() const { + return schema_json_; + } + const RestAuditFields& GetAuditFields() const { + return audit_; + } + + GetTableResponse() = default; + + private: + std::string id_; + std::string name_; + std::string path_; + bool is_external_ = false; + int64_t schema_id_ = 0; + std::string schema_json_; + RestAuditFields audit_; +}; + +/// `schema_json` uses the Java `org.apache.paimon.schema.Schema` JSON layout. +class CreateTableRequest : public Jsonizable { + public: + CreateTableRequest(const std::string& database, const std::string& table, + const std::string& schema_json) + : database_(database), table_(table), schema_json_(schema_json) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::string& GetDatabase() const { + return database_; + } + const std::string& GetTable() const { + return table_; + } + const std::string& GetSchemaJson() const { + return schema_json_; + } + + CreateTableRequest() = default; + + private: + std::string database_; + std::string table_; + std::string schema_json_; +}; + +class RenameTableRequest : public Jsonizable { + public: + RenameTableRequest(const std::string& source_database, const std::string& source_table, + const std::string& destination_database, + const std::string& destination_table) + : source_database_(source_database), + source_table_(source_table), + destination_database_(destination_database), + destination_table_(destination_table) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::string& GetSourceDatabase() const { + return source_database_; + } + const std::string& GetSourceTable() const { + return source_table_; + } + const std::string& GetDestinationDatabase() const { + return destination_database_; + } + const std::string& GetDestinationTable() const { + return destination_table_; + } + + RenameTableRequest() = default; + + private: + std::string source_database_; + std::string source_table_; + std::string destination_database_; + std::string destination_table_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/rest_messages_test.cpp b/src/paimon/rest/rest_messages_test.cpp new file mode 100644 index 000000000..1b8313162 --- /dev/null +++ b/src/paimon/rest/rest_messages_test.cpp @@ -0,0 +1,281 @@ +/* + * 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/rest/rest_messages.h" + +#include +#include + +#include "gtest/gtest.h" +#include "paimon/rest/resource_paths.h" +#include "paimon/rest/rest_util.h" +#include "paimon/testing/utils/testharness.h" +#include "rapidjson/document.h" + +namespace paimon::test { + +TEST(RestUtilTest, EncodeString) { + ASSERT_EQ("abcDEF012.-*_", RestUtil::EncodeString("abcDEF012.-*_")); + ASSERT_EQ("a+b", RestUtil::EncodeString("a b")); + ASSERT_EQ("a%2Fb", RestUtil::EncodeString("a/b")); + ASSERT_EQ("a%3Db%26c", RestUtil::EncodeString("a=b&c")); + ASSERT_EQ("%E4%B8%AD", RestUtil::EncodeString("中")); +} + +TEST(RestUtilTest, DecodeString) { + ASSERT_EQ("a/b", RestUtil::DecodeString("a%2Fb")); + ASSERT_EQ("a b", RestUtil::DecodeString("a+b")); + ASSERT_EQ("中", RestUtil::DecodeString("%E4%B8%AD")); + // an escape ending the input is still decoded + ASSERT_EQ("a/", RestUtil::DecodeString("a%2F")); + // malformed escapes are kept as-is instead of failing + ASSERT_EQ("a%", RestUtil::DecodeString("a%")); + ASSERT_EQ("%2", RestUtil::DecodeString("%2")); + ASSERT_EQ("%ZZ", RestUtil::DecodeString("%ZZ")); + ASSERT_EQ("100%", RestUtil::DecodeString("100%")); +} + +TEST(RestUtilTest, ExtractPrefixMap) { + std::map options = { + {"header.k1", "v1"}, {"header.k2", "v2"}, {"other", "v3"}, {"header.", "v4"}}; + std::map expected = {{"k1", "v1"}, {"k2", "v2"}}; + ASSERT_EQ(expected, RestUtil::ExtractPrefixMap(options, "header.")); +} + +TEST(ResourcePathsTest, WithPrefix) { + ResourcePaths paths("my prefix"); + ASSERT_EQ("/v1/config", ResourcePaths::Config()); + ASSERT_EQ("/v1/my+prefix/databases", paths.Databases()); + ASSERT_EQ("/v1/my+prefix/databases/db%231", paths.Database("db#1")); + ASSERT_EQ("/v1/my+prefix/databases/db/tables", paths.Tables("db")); + ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1", paths.Table("db", "t1")); + ASSERT_EQ("/v1/my+prefix/tables/rename", paths.RenameTable()); + ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1/snapshots", paths.Snapshots("db", "t1")); +} + +TEST(ResourcePathsTest, WithoutPrefix) { + ResourcePaths paths(""); + ASSERT_EQ("/v1/databases", paths.Databases()); + ASSERT_EQ("/v1/tables/rename", paths.RenameTable()); +} + +TEST(RestMessagesTest, ErrorResponseRoundTrip) { + ErrorResponse response(ErrorResponse::kResourceTypeDatabase, "db1", "database db1 not found", + 404); + ASSERT_OK_AND_ASSIGN(std::string json, response.ToJsonString()); + ASSERT_OK_AND_ASSIGN(ErrorResponse parsed, ErrorResponse::FromJsonString(json)); + ASSERT_EQ("DATABASE", parsed.GetResourceType()); + ASSERT_EQ("db1", parsed.GetResourceName()); + ASSERT_EQ("database db1 not found", parsed.GetMessage()); + ASSERT_EQ(404, parsed.GetCode()); +} + +TEST(RestMessagesTest, ErrorResponseLenientParse) { + ASSERT_OK_AND_ASSIGN(ErrorResponse parsed, ErrorResponse::FromJsonString("{}")); + ASSERT_EQ("", parsed.GetMessage()); + ASSERT_EQ(0, parsed.GetCode()); +} + +TEST(RestMessagesTest, ConfigResponseMerge) { + // null values may be sent by the Java server and are filtered out + std::string json = R"({ + "defaults": {"prefix": "server-prefix", "a": "default-a", "b": "default-b", + "nullable": null}, + "overrides": {"c": "override-c", "a": "override-a"} + })"; + ASSERT_OK_AND_ASSIGN(ConfigResponse config, ConfigResponse::FromJsonString(json)); + std::map client = { + {"a", "client-a"}, {"b", "client-b"}, {"d", "client-d"}}; + std::map merged = config.Merge(client); + // overrides > client options > defaults + ASSERT_EQ("override-a", merged["a"]); + ASSERT_EQ("client-b", merged["b"]); + ASSERT_EQ("override-c", merged["c"]); + ASSERT_EQ("client-d", merged["d"]); + ASSERT_EQ("server-prefix", merged["prefix"]); + ASSERT_EQ(0, merged.count("nullable")); +} + +TEST(RestMessagesTest, ListResponsesParse) { + ASSERT_OK_AND_ASSIGN(ListDatabasesResponse databases, + ListDatabasesResponse::FromJsonString( + R"({"databases": ["db1", "db2"], "nextPageToken": "token1"})")); + ASSERT_EQ((std::vector{"db1", "db2"}), databases.Data()); + ASSERT_EQ("token1", databases.NextPageToken().value()); + + ASSERT_OK_AND_ASSIGN(ListTablesResponse tables, + ListTablesResponse::FromJsonString(R"({"tables": ["t1"]})")); + ASSERT_EQ((std::vector{"t1"}), tables.Data()); + ASSERT_FALSE(tables.NextPageToken().has_value()); + + // nextPageToken serialized as explicit null by the Java server + ASSERT_OK_AND_ASSIGN( + ListTablesResponse null_token, + ListTablesResponse::FromJsonString(R"({"tables": [], "nextPageToken": null})")); + ASSERT_FALSE(null_token.NextPageToken().has_value()); +} + +TEST(RestMessagesTest, ListSnapshotsResponseParse) { + std::string json = R"({ + "snapshots": [{ + "version": 3, + "id": 7, + "schemaId": 2, + "baseManifestList": "manifest-list-1", + "deltaManifestList": "manifest-list-2", + "commitUser": "user1", + "commitIdentifier": 9, + "commitKind": "APPEND", + "timeMillis": 1234567, + "totalRecordCount": 100, + "deltaRecordCount": 10, + "watermark": 42 + }, { + "version": 3, + "id": 8, + "schemaId": 2, + "baseManifestList": "manifest-list-3", + "deltaManifestList": "manifest-list-4", + "commitUser": "user1", + "commitIdentifier": 10, + "commitKind": "COMPACT", + "timeMillis": 1234568, + "totalRecordCount": 100, + "deltaRecordCount": 0 + }], + "nextPageToken": null + })"; + ASSERT_OK_AND_ASSIGN(ListSnapshotsResponse response, + ListSnapshotsResponse::FromJsonString(json)); + ASSERT_EQ(2, response.Data().size()); + const Snapshot& snapshot = response.Data()[0]; + ASSERT_EQ(7, snapshot.Id()); + ASSERT_EQ(2, snapshot.SchemaId()); + ASSERT_EQ("user1", snapshot.CommitUser()); + ASSERT_EQ(1234567, snapshot.TimeMillis()); + SnapshotInfo info = snapshot.ToSnapshotInfo(); + ASSERT_EQ(SnapshotInfo::CommitKind::APPEND, info.commit_kind); + ASSERT_EQ(100, info.total_record_count.value()); + ASSERT_EQ(42, info.watermark.value()); + SnapshotInfo compact_info = response.Data()[1].ToSnapshotInfo(); + ASSERT_EQ(SnapshotInfo::CommitKind::COMPACT, compact_info.commit_kind); + ASSERT_FALSE(compact_info.watermark.has_value()); +} + +TEST(RestMessagesTest, GetDatabaseResponseParse) { + std::string json = R"({ + "id": "10", + "name": "db1", + "location": "/warehouse/db1.db", + "options": {"k1": "v1"}, + "owner": "owner1", + "createdAt": 100, + "createdBy": "creator", + "updatedAt": 200, + "updatedBy": "updater" + })"; + ASSERT_OK_AND_ASSIGN(GetDatabaseResponse response, GetDatabaseResponse::FromJsonString(json)); + ASSERT_EQ("db1", response.GetName()); + ASSERT_EQ("/warehouse/db1.db", response.GetLocation()); + ASSERT_EQ("v1", response.GetOptions().at("k1")); + std::map options; + response.GetAuditFields().PutAuditOptionsTo(&options); + ASSERT_EQ("owner1", options.at("owner")); + ASSERT_EQ("100", options.at("createdAt")); + ASSERT_EQ("updater", options.at("updatedBy")); +} + +TEST(RestMessagesTest, GetTableResponseParse) { + std::string json = R"({ + "id": "42", + "name": "t1", + "path": "/warehouse/db1.db/t1", + "isExternal": false, + "schemaId": 3, + "schema": { + "fields": [ + {"id": 0, "name": "f0", "type": "INT NOT NULL"}, + {"id": 1, "name": "f1", "type": "STRING"} + ], + "partitionKeys": [], + "primaryKeys": ["f0"], + "options": {"bucket": "2"}, + "comment": "a table" + }, + "updatedAt": 300 + })"; + ASSERT_OK_AND_ASSIGN(GetTableResponse response, GetTableResponse::FromJsonString(json)); + ASSERT_EQ("42", response.GetId()); + ASSERT_EQ("t1", response.GetName()); + ASSERT_EQ("/warehouse/db1.db/t1", response.GetPath()); + ASSERT_FALSE(response.IsExternal()); + ASSERT_EQ(3, response.GetSchemaId()); + ASSERT_EQ(300, response.GetAuditFields().updated_at.value()); + rapidjson::Document schema; + schema.Parse(response.GetSchemaJson().c_str()); + ASSERT_FALSE(schema.HasParseError()); + ASSERT_EQ(2u, schema["fields"].Size()); + ASSERT_STREQ("f0", schema["fields"][0]["name"].GetString()); +} + +TEST(RestMessagesTest, CreateTableRequestSerialize) { + CreateTableRequest request( + "db1", "t1", R"({"fields": [], "partitionKeys": [], "primaryKeys": [], "options": {}})"); + ASSERT_OK_AND_ASSIGN(std::string json, request.ToJsonString()); + rapidjson::Document doc; + doc.Parse(json.c_str()); + ASSERT_FALSE(doc.HasParseError()); + ASSERT_STREQ("db1", doc["identifier"]["database"].GetString()); + ASSERT_STREQ("t1", doc["identifier"]["object"].GetString()); + ASSERT_TRUE(doc["schema"].IsObject()); + ASSERT_TRUE(doc["schema"]["fields"].IsArray()); + + ASSERT_OK_AND_ASSIGN(CreateTableRequest parsed, CreateTableRequest::FromJsonString(json)); + ASSERT_EQ("db1", parsed.GetDatabase()); + ASSERT_EQ("t1", parsed.GetTable()); +} + +TEST(RestMessagesTest, InvalidSchemaJsonErrorOmitsPayload) { + CreateTableRequest request("db1", "t1", R"({"fields": [], "options": {"token": "top-secret")"); + Status status = request.ToJsonString().status(); + ASSERT_NOK_WITH_MSG(status, "invalid json"); + // the payload may carry credentials, so it must not be echoed back in the error + ASSERT_EQ(std::string::npos, status.ToString().find("top-secret")) << status.ToString(); +} + +TEST(RestMessagesTest, RenameTableRequestSerialize) { + RenameTableRequest request("db1", "t1", "db1", "t2"); + ASSERT_OK_AND_ASSIGN(std::string json, request.ToJsonString()); + rapidjson::Document doc; + doc.Parse(json.c_str()); + ASSERT_FALSE(doc.HasParseError()); + ASSERT_STREQ("t1", doc["source"]["object"].GetString()); + ASSERT_STREQ("t2", doc["destination"]["object"].GetString()); + + ASSERT_OK_AND_ASSIGN(RenameTableRequest parsed, RenameTableRequest::FromJsonString(json)); + ASSERT_EQ("db1", parsed.GetSourceDatabase()); + ASSERT_EQ("t2", parsed.GetDestinationTable()); +} + +TEST(RestMessagesTest, CreateDatabaseRequestRoundTrip) { + CreateDatabaseRequest request("db1", {{"k1", "v1"}}); + ASSERT_OK_AND_ASSIGN(std::string json, request.ToJsonString()); + ASSERT_OK_AND_ASSIGN(CreateDatabaseRequest parsed, CreateDatabaseRequest::FromJsonString(json)); + ASSERT_EQ("db1", parsed.GetName()); + ASSERT_EQ("v1", parsed.GetOptions().at("k1")); +} + +} // namespace paimon::test diff --git a/src/paimon/rest/rest_util.cpp b/src/paimon/rest/rest_util.cpp new file mode 100644 index 000000000..63b0a84d4 --- /dev/null +++ b/src/paimon/rest/rest_util.cpp @@ -0,0 +1,122 @@ +/* + * 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/rest/rest_util.h" + +#include +#include + +#include "fmt/format.h" +#include "rapidjson/error/en.h" +#include "rapidjson/stringbuffer.h" +#include "rapidjson/writer.h" + +namespace paimon { + +namespace { +bool IsUnreservedChar(char c) { + return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '.' || + c == '-' || c == '*' || c == '_'; +} + +int32_t HexValue(char c) { + if (c >= '0' && c <= '9') { + return c - '0'; + } + if (c >= 'a' && c <= 'f') { + return c - 'a' + 10; + } + if (c >= 'A' && c <= 'F') { + return c - 'A' + 10; + } + return -1; +} +} // namespace + +std::string RestUtil::EncodeString(const std::string& input) { + std::string encoded; + encoded.reserve(input.size()); + for (char c : input) { + if (IsUnreservedChar(c)) { + encoded.push_back(c); + } else if (c == ' ') { + encoded.push_back('+'); + } else { + char buf[4]; + std::snprintf(buf, sizeof(buf), "%%%02X", static_cast(c)); + encoded.append(buf); + } + } + return encoded; +} + +std::string RestUtil::DecodeString(const std::string& input) { + std::string decoded; + decoded.reserve(input.size()); + for (size_t i = 0; i < input.size(); i++) { + char c = input[i]; + if (c == '+') { + decoded.push_back(' '); + } else if (c == '%' && i + 2 < input.size()) { + int32_t high = HexValue(input[i + 1]); + int32_t low = HexValue(input[i + 2]); + if (high >= 0 && low >= 0) { + decoded.push_back(static_cast((high << 4) | low)); + i += 2; + } else { + decoded.push_back(c); + } + } else { + decoded.push_back(c); + } + } + return decoded; +} + +std::map RestUtil::ExtractPrefixMap( + const std::map& options, const std::string& prefix) { + std::map result; + for (const auto& [key, value] : options) { + if (key.size() > prefix.size() && key.compare(0, prefix.size(), prefix) == 0) { + result[key.substr(prefix.size())] = value; + } + } + return result; +} + +std::string RestUtil::JsonToString(const rapidjson::Value& value) { + rapidjson::StringBuffer buffer; + rapidjson::Writer writer(buffer); + value.Accept(writer); + return buffer.GetString(); +} + +rapidjson::Value RestUtil::ParseToValue(const std::string& json, + rapidjson::Document::AllocatorType* allocator) { + rapidjson::Document doc; + doc.Parse(json.c_str()); + if (doc.HasParseError()) { + // The payload may carry credentials and be arbitrarily large; report only the error. + throw std::invalid_argument(fmt::format("invalid json: {} (at offset {})", + rapidjson::GetParseError_En(doc.GetParseError()), + doc.GetErrorOffset())); + } + rapidjson::Value value; + value.CopyFrom(doc, *allocator); + return value; +} + +} // namespace paimon diff --git a/src/paimon/rest/rest_util.h b/src/paimon/rest/rest_util.h new file mode 100644 index 000000000..dc3eb6447 --- /dev/null +++ b/src/paimon/rest/rest_util.h @@ -0,0 +1,57 @@ +/* + * 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 +#include + +#include "rapidjson/allocators.h" +#include "rapidjson/document.h" +#include "rapidjson/rapidjson.h" + +namespace paimon { + +/// Utilities for the REST catalog, mirroring `RESTUtil` in the Java implementation. +class RestUtil { + public: + RestUtil() = delete; + ~RestUtil() = delete; + + /// URL-encode a string with the same rules as `java.net.URLEncoder#encode` in UTF-8: + /// alphanumeric characters and ".", "-", "*", "_" are kept, a space becomes "+", all + /// other bytes become percent-encoded "%XX". + static std::string EncodeString(const std::string& input); + + /// Decodes `EncodeString` output ('+' back to space, "%XX" to the byte); malformed + /// escape sequences are kept as-is instead of failing. + static std::string DecodeString(const std::string& input); + + /// Extract all options whose key starts with `prefix`, with the prefix stripped from + /// the resulting keys. + static std::map ExtractPrefixMap( + const std::map& options, const std::string& prefix); + + /// Serialize a rapidjson value to a compact JSON string. + static std::string JsonToString(const rapidjson::Value& value); + + /// Parse a JSON string into a rapidjson value owned by `allocator`. + /// Throws `std::invalid_argument` on parse error, consistent with `Jsonizable`. + static rapidjson::Value ParseToValue(const std::string& json, + rapidjson::Document::AllocatorType* allocator); +}; + +} // namespace paimon