From 1c3e86edc413e5fed15a5f3f547c2a70bb72405d Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Tue, 27 Oct 2026 14:47:39 +0800 Subject: [PATCH] feat(rest): support table commit leases Port REST catalog table commit leases from Java Paimon. The table property rest.commit.lock-enabled turns on server-side lease enforcement, and a writer that sets rest.commit.acquire-lock acquires a lease for each publication attempt. Both default to false. Asking to acquire without the table capability, without a table UUID, or in a request-only commit built with UseRESTCatalogCommit() fails when the commit is created. Add an internal CommitLock/CommitLockScope interface mirroring Java Lock. SnapshotCommit exposes AcquireLock(), CatalogSnapshotCommit holds the lock created by VersionManagedCatalog::CreateCommitLock(), and RestCatalog returns a RestCommitLock bound to the branch identifier, table UUID and commit user. RestCommitLock acquires a lease through the commit-lock endpoint. While another writer holds it, acquisition logs that it is waiting, then polls until commit.timeout without using up commit.max-retries; the timeout is checked after each response rather than interrupting a request in flight. A grant for another commit user or without a positive lease duration is rejected with an error naming the expected and actual value, and a renewal thread that cannot be started fails the acquisition instead of throwing. Each scope renews its lease through commit-lock/renew every third of the lease duration and tracks the deadline on a monotonic clock from when each grant was requested, so a refused or failed renewal, a changed lease duration or a local expiry invalidates it. There is no unlock request: ending a scope only stops renewing, and never waits for a renewal in flight, even one refreshing credentials, since renewal runs on a detached thread sharing the lease state. The renewal request is cancelled through a new RestHttpClient::Cancellation, which aborts the libcurl transfer and ends a retry backoff. FileStoreCommitImpl acquires a scope before reading the head of each attempt, checks it in CommitSnapshotImpl right before SnapshotCommit::Commit(), and releases it before waiting to retry. A failed acquisition, or a scope found invalid, publishes nothing and is retried within the commit.max-retries and commit.timeout budget of a conflict. An attempt that lost its scope removes the manifests it wrote, as one that lost a race does, and that cleanup now also covers changelog manifests and their list. Running out of retries reports the lease failure instead of a conflict. RollbackToAsLatest holds a scope too; failing to acquire or losing it fails the rollback, and a lost scope also removes the manifests it wrote. --- docs/source/user_guide/catalog.rst | 52 ++ include/paimon/defs.h | 10 + src/paimon/CMakeLists.txt | 2 + src/paimon/common/defs.cpp | 2 + .../core/catalog/catalog_snapshot_commit.h | 16 +- .../catalog/catalog_snapshot_commit_test.cpp | 44 ++ src/paimon/core/catalog/commit_lock.h | 49 ++ src/paimon/core/catalog/snapshot_commit.h | 10 + .../core/catalog/version_managed_catalog.h | 16 + src/paimon/core/core_options.cpp | 16 + src/paimon/core/core_options.h | 2 + src/paimon/core/core_options_test.cpp | 6 + .../core/operation/file_store_commit_impl.cpp | 141 +++- .../core/operation/file_store_commit_impl.h | 23 +- .../operation/file_store_commit_impl_test.cpp | 252 +++++++- src/paimon/core/table/table.cpp | 23 +- src/paimon/rest/resource_paths.cpp | 10 + src/paimon/rest/resource_paths.h | 5 + src/paimon/rest/resource_paths_test.cpp | 5 + src/paimon/rest/rest_api.cpp | 37 +- src/paimon/rest/rest_api.h | 37 +- src/paimon/rest/rest_catalog.cpp | 20 + src/paimon/rest/rest_catalog.h | 7 + src/paimon/rest/rest_catalog_test.cpp | 361 ++++++++++- src/paimon/rest/rest_commit_lock.cpp | 286 +++++++++ src/paimon/rest/rest_commit_lock.h | 99 +++ src/paimon/rest/rest_commit_lock_test.cpp | 600 ++++++++++++++++++ src/paimon/rest/rest_http_client.cpp | 57 +- src/paimon/rest/rest_http_client.h | 31 +- src/paimon/rest/rest_http_client_test.cpp | 122 ++++ src/paimon/rest/rest_messages.cpp | 45 ++ src/paimon/rest/rest_messages.h | 74 +++ src/paimon/rest/rest_messages_test.cpp | 69 ++ src/paimon/testing/mock/mock_catalog.h | 18 + 34 files changed, 2494 insertions(+), 53 deletions(-) create mode 100644 src/paimon/core/catalog/commit_lock.h create mode 100644 src/paimon/rest/rest_commit_lock.cpp create mode 100644 src/paimon/rest/rest_commit_lock.h create mode 100644 src/paimon/rest/rest_commit_lock_test.cpp diff --git a/docs/source/user_guide/catalog.rst b/docs/source/user_guide/catalog.rst index e311b52b0..2d789f839 100644 --- a/docs/source/user_guide/catalog.rst +++ b/docs/source/user_guide/catalog.rst @@ -197,6 +197,58 @@ can tell a commit that raced another. A commit which lost such a race is rebased on the new latest snapshot and retried, bounded by ``commit.max-retries`` and ``commit.timeout`` as a file-system commit is. +A REST catalog can also give a writer a commit lease, so concurrent writers take +turns instead of racing. The table property ``rest.commit.lock-enabled = true`` +turns on lease enforcement on the server, and a writer that sets +``rest.commit.acquire-lock = true``, usually as a commit context option, acquires +a lease for each publication attempt. Both default to ``false``; enabling the +table property alone does not make writers acquire, and asking to acquire without +it fails when the commit is created. A writer with an active lease keeps the +others out, including writers that do not acquire one. The lease is bound to the +table or branch, to the table UUID, which has to be passed to ``WithTableId``, +and to the exact commit user, so give each logical writer its own commit user and +keep it across recovery. + +.. code-block:: cpp + + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr table, + shared_catalog->GetTable(paimon::Identifier("db", "tbl"))); + std::optional table_uuid = table->CatalogUuid(); + if (!table_uuid) { + return paimon::Status::Invalid("a commit lease needs the table UUID"); + } + paimon::CommitContextBuilder builder(table_path, "writer-1"); + builder.WithCatalog(shared_catalog, paimon::Identifier("db", "tbl")) + .WithTableId(table_uuid.value()) + .AddOption(paimon::Options::REST_COMMIT_ACQUIRE_LOCK, "true"); + +Each attempt acquires its own lease before reading the latest snapshot, waits +while another writer holds it until ``commit.timeout`` without using up +``commit.max-retries``, renews it in the background and checks it is still +valid right before publishing. An attempt that fails to acquire the lease, or +loses it before publishing, publishes nothing and is retried within +``commit.max-retries`` and ``commit.timeout``, as after a conflict; an attempt +that lost its lease also removes the manifests it wrote. When the retries run +out, the error names the lease failure rather than a conflict. +``RollbackToAsLatest`` holds a lease the same way but is not retried: failing to +acquire or losing the lease fails it. + +``commit.timeout`` is not a strict deadline for waiting on the lease. Each +acquisition may wait the full timeout, however much of it earlier attempts of +the same commit used. The timeout is also checked only after each acquisition +request returns, without interrupting a request in flight, so the wait can run +past it by the duration of one request, which only the HTTP client's own +timeouts bound. + +There is no unlock request. The server ends a lease when a snapshot is published +under it. Ending an attempt that published nothing only stops renewing the +lease, without waiting for a renewal in flight, and the lease stays held until +it expires. + +Leases need ``WithCatalog``: a commit that only builds the commit table request +for the caller to send cannot hold one, so asking to acquire is rejected there. +The snapshot UUID checks above still apply. + The HTTP client neither retries nor follows redirects for a commit request. Transport errors and non-success responses, including 429, 503 and redirects, leave the outcome uncertain: the server may already have accepted the snapshot. diff --git a/include/paimon/defs.h b/include/paimon/defs.h index 205bc5eda..55fd956d8 100644 --- a/include/paimon/defs.h +++ b/include/paimon/defs.h @@ -314,6 +314,16 @@ struct PAIMON_EXPORT Options { /// Default value is "false". static const char COMMIT_DISCARD_DUPLICATE_FILES[]; + /// "rest.commit.lock-enabled" - Enable the REST catalog's server-side table commit lease + /// capability. Writers acquire a lease only when "rest.commit.acquire-lock" is enabled. + /// Default value is "false". + static const char REST_COMMIT_LOCK_ENABLED[]; + + /// "rest.commit.acquire-lock" - Whether this writer acquires a REST commit lease for each + /// publication attempt. Can be set dynamically for a writer and requires + /// "rest.commit.lock-enabled" on the table. Default value is "false". + static const char REST_COMMIT_ACQUIRE_LOCK[]; + /// "compaction.max-size-amplification-percent" - The size amplification is defined as the /// amount (in percentage) of additional storage needed to store a single byte of data in the /// merge tree for changelog mode table. Default value is 200. diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index f92450d39..b4ad6e80f 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -514,6 +514,7 @@ if(PAIMON_ENABLE_REST) rest/rest_auth.cpp rest/dlf_auth.cpp rest/rest_catalog.cpp + rest/rest_commit_lock.cpp rest/rest_messages.cpp rest/rest_token_file_system.cpp rest/rest_credential_provider.cpp @@ -1100,6 +1101,7 @@ if(PAIMON_BUILD_TESTS) rest/resource_paths_test.cpp rest/dlf_auth_test.cpp rest/rest_catalog_test.cpp + rest/rest_commit_lock_test.cpp rest/rest_messages_test.cpp rest/rest_token_file_system_test.cpp rest/rest_credential_provider_test.cpp diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp index 5239ec19b..753dbb43d 100644 --- a/src/paimon/common/defs.cpp +++ b/src/paimon/common/defs.cpp @@ -89,6 +89,8 @@ const char Options::COMMIT_MAX_RETRIES[] = "commit.max-retries"; const char Options::COMMIT_MIN_RETRY_WAIT[] = "commit.min-retry-wait"; const char Options::COMMIT_MAX_RETRY_WAIT[] = "commit.max-retry-wait"; const char Options::COMMIT_DISCARD_DUPLICATE_FILES[] = "commit.discard-duplicate-files"; +const char Options::REST_COMMIT_LOCK_ENABLED[] = "rest.commit.lock-enabled"; +const char Options::REST_COMMIT_ACQUIRE_LOCK[] = "rest.commit.acquire-lock"; const char Options::SEQUENCE_FIELD[] = "sequence.field"; const char Options::SEQUENCE_FIELD_SORT_ORDER[] = "sequence.field.sort-order"; const char Options::MERGE_ENGINE[] = "merge-engine"; diff --git a/src/paimon/core/catalog/catalog_snapshot_commit.h b/src/paimon/core/catalog/catalog_snapshot_commit.h index 0c47c99fb..f9b07cf1e 100644 --- a/src/paimon/core/catalog/catalog_snapshot_commit.h +++ b/src/paimon/core/catalog/catalog_snapshot_commit.h @@ -27,6 +27,7 @@ #include "paimon/catalog/catalog.h" #include "paimon/catalog/identifier.h" #include "paimon/core/catalog/catalog_utils.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/catalog/commit_table_request.h" #include "paimon/core/catalog/snapshot_commit.h" #include "paimon/core/catalog/version_managed_catalog.h" @@ -45,17 +46,27 @@ class CatalogSnapshotCommit : public SnapshotCommit { /// @param catalog Catalog implementing version management. /// @param identifier Identifier of the table to commit to. /// @param table_id Catalog UUID captured before preparing changes; null if unavailable. + /// @param lock Lock taken around each publication attempt; null to publish without one. CatalogSnapshotCommit(const std::shared_ptr& catalog, Identifier identifier, - const std::optional& table_id) + const std::optional& table_id, + std::unique_ptr lock = nullptr) : catalog_(catalog), version_managed_catalog_(AsVersionManaged(catalog)), identifier_(std::move(identifier)), - table_id_(table_id) {} + table_id_(table_id), + lock_(std::move(lock)) {} /// @param table_id Catalog UUID to include in the request; null if unavailable. explicit CatalogSnapshotCommit(const std::optional& table_id = std::nullopt) : identifier_("", ""), table_id_(table_id) {} + Result> AcquireLock() override { + if (lock_ == nullptr) { + return std::unique_ptr(); + } + return lock_->Acquire(); + } + Result Commit(const std::optional& base_snapshot_uuid, const Snapshot& snapshot, const std::string& branch, const std::vector& statistics) override { @@ -110,6 +121,7 @@ class CatalogSnapshotCommit : public SnapshotCommit { VersionManagedCatalog* version_managed_catalog_ = nullptr; Identifier identifier_; std::optional table_id_; + std::unique_ptr lock_; std::optional commit_table_request_; }; diff --git a/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp b/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp index b062c0027..03a274135 100644 --- a/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp +++ b/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp @@ -26,6 +26,7 @@ #include "gtest/gtest.h" #include "paimon/catalog/identifier.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/catalog/commit_table_request.h" #include "paimon/core/partition/partition_statistics.h" #include "paimon/result.h" @@ -191,4 +192,47 @@ TEST(CatalogSnapshotCommitTest, TestBuildRequestWithoutCatalog) { ASSERT_EQ(bare_request.GetBaseSnapshotUuid(), std::nullopt); } +TEST(CatalogSnapshotCommitTest, TestAcquireLockUsesTheLockItWasGiven) { + class CountingLock : public CommitLock { + public: + explicit CountingLock(int32_t* acquisitions) : acquisitions_(acquisitions) {} + + Result> Acquire() override { + ++*acquisitions_; + if (*acquisitions_ > 1) { + return Status::IOError("lock service unavailable"); + } + return std::unique_ptr(std::make_unique()); + } + + private: + class Scope : public CommitLockScope { + public: + Status EnsureValid() override { + return Status::OK(); + } + }; + + int32_t* acquisitions_; + }; + + auto catalog = std::make_shared(true); + CatalogSnapshotCommit unlocked(catalog, Identifier("db", "tbl"), "table-uuid"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, unlocked.AcquireLock()); + ASSERT_FALSE(scope); + CatalogSnapshotCommit request_only("table-uuid"); + ASSERT_OK_AND_ASSIGN(scope, request_only.AcquireLock()); + ASSERT_FALSE(scope); + + int32_t acquisitions = 0; + CatalogSnapshotCommit locked(catalog, Identifier("db", "tbl"), "table-uuid", + std::make_unique(&acquisitions)); + ASSERT_OK_AND_ASSIGN(scope, locked.AcquireLock()); + ASSERT_TRUE(scope); + ASSERT_OK(scope->EnsureValid()); + ASSERT_NOK_WITH_MSG(locked.AcquireLock(), "lock service unavailable"); + ASSERT_EQ(2, acquisitions); + ASSERT_TRUE(catalog->CommitCalls().empty()); +} + } // namespace paimon::test diff --git a/src/paimon/core/catalog/commit_lock.h b/src/paimon/core/catalog/commit_lock.h new file mode 100644 index 000000000..5619a64a4 --- /dev/null +++ b/src/paimon/core/catalog/commit_lock.h @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +/// One acquisition of a `CommitLock`, held for a single snapshot publication attempt. Destroying +/// the scope ends it. +class CommitLockScope { + public: + virtual ~CommitLockScope() = default; + + /// Returns an error once the scope can no longer protect a publication, for example after its + /// lease expired. Called right before the snapshot is published. + virtual Status EnsureValid() = 0; +}; + +/// A lock taken around each snapshot publication attempt, mirroring Java +/// `org.apache.paimon.operation.Lock`. +class CommitLock { + public: + virtual ~CommitLock() = default; + + /// Acquires a scope for one publication attempt, waiting while another writer holds the lock. + virtual Result> Acquire() = 0; +}; + +} // namespace paimon diff --git a/src/paimon/core/catalog/snapshot_commit.h b/src/paimon/core/catalog/snapshot_commit.h index 9500205cb..41bd1712a 100644 --- a/src/paimon/core/catalog/snapshot_commit.h +++ b/src/paimon/core/catalog/snapshot_commit.h @@ -18,10 +18,12 @@ #pragma once +#include #include #include #include +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/partition/partition_statistics.h" #include "paimon/core/snapshot.h" @@ -32,6 +34,14 @@ class SnapshotCommit { public: virtual ~SnapshotCommit() = default; + /// Acquires the lock scope of one publication attempt. The caller holds it from reading the + /// base snapshot until `Commit()` returns, and checks it right before committing. + /// + /// @return The scope, or null when publication takes no lock. + virtual Result> AcquireLock() { + return std::unique_ptr(); + } + /// @param base_snapshot_uuid Base snapshot UUID; null for an absent or legacy snapshot. /// @param snapshot Snapshot to be committed. /// @param branch Branch to publish the snapshot on; an empty name is the main branch. diff --git a/src/paimon/core/catalog/version_managed_catalog.h b/src/paimon/core/catalog/version_managed_catalog.h index 16c4aeb6a..7812e6617 100644 --- a/src/paimon/core/catalog/version_managed_catalog.h +++ b/src/paimon/core/catalog/version_managed_catalog.h @@ -25,12 +25,14 @@ #include "paimon/catalog/catalog.h" #include "paimon/catalog/identifier.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/partition/partition_statistics.h" #include "paimon/core/snapshot.h" #include "paimon/result.h" #include "paimon/status.h" namespace paimon { +class CoreOptions; /// Snapshot publication interface for catalogs returning true from `SupportsVersionManagement()`. /// @@ -59,6 +61,20 @@ class VersionManagedCatalog { const std::optional& base_snapshot_uuid, const Snapshot& snapshot, const std::vector& statistics) = 0; + + /// Creates the lock taken around each snapshot publication to a table or branch, mirroring + /// Java `Catalog.createLock`. + /// + /// @param identifier Identifier of the table or branch the snapshots are published to. + /// @param table_uuid Catalog UUID of the table or branch; null if unavailable. + /// @param commit_user Commit user of the snapshots published under the lock. + /// @param options Table options merged with the writer's dynamic options. + /// @return The lock, or null when publication takes none. + virtual Result> CreateCommitLock( + const Identifier& identifier, const std::optional& table_uuid, + const std::string& commit_user, const CoreOptions& options) { + return std::unique_ptr(); + } }; /// Returns version management only when the catalog both advertises and implements it. diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp index d342ee826..c5018f6aa 100644 --- a/src/paimon/core/core_options.cpp +++ b/src/paimon/core/core_options.cpp @@ -474,6 +474,8 @@ struct CoreOptions::Impl { bool global_index_enabled = true; bool commit_force_compact = false; bool commit_discard_duplicate_files = false; + bool rest_commit_lock_enabled = false; + bool rest_commit_acquire_lock = false; bool dynamic_partition_overwrite = true; bool overwrite_upgrade = true; bool compaction_force_rewrite_all_files = false; @@ -704,6 +706,12 @@ struct CoreOptions::Impl { // Parse commit.discard-duplicate-files - whether to discard duplicate files on append PAIMON_RETURN_NOT_OK(parser.Parse(Options::COMMIT_DISCARD_DUPLICATE_FILES, &commit_discard_duplicate_files)); + // Parse rest.commit.lock-enabled - server-side commit lease capability of the table + PAIMON_RETURN_NOT_OK( + parser.Parse(Options::REST_COMMIT_LOCK_ENABLED, &rest_commit_lock_enabled)); + // Parse rest.commit.acquire-lock - whether this writer acquires a commit lease + PAIMON_RETURN_NOT_OK( + parser.Parse(Options::REST_COMMIT_ACQUIRE_LOCK, &rest_commit_acquire_lock)); // Parse dynamic-partition-overwrite - whether overwrite only dynamic partitions // for partitioned table overwrite. PAIMON_RETURN_NOT_OK( @@ -1343,6 +1351,14 @@ bool CoreOptions::CommitDiscardDuplicateFiles() const { return impl_->commit_discard_duplicate_files; } +bool CoreOptions::RestCommitLockEnabled() const { + return impl_->rest_commit_lock_enabled; +} + +bool CoreOptions::RestCommitAcquireLock() const { + return impl_->rest_commit_acquire_lock; +} + bool CoreOptions::DynamicPartitionOverwrite() const { return impl_->dynamic_partition_overwrite; } diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h index 474bfe136..61562deb8 100644 --- a/src/paimon/core/core_options.h +++ b/src/paimon/core/core_options.h @@ -142,6 +142,8 @@ class PAIMON_EXPORT CoreOptions { int64_t GetCommitMinRetryWait() const; int64_t GetCommitMaxRetryWait() const; bool CommitDiscardDuplicateFiles() const; + bool RestCommitLockEnabled() const; + bool RestCommitAcquireLock() const; bool DynamicPartitionOverwrite() const; bool OverwriteUpgrade() const; int32_t GetCompactionMinFileNum() const; diff --git a/src/paimon/core/core_options_test.cpp b/src/paimon/core/core_options_test.cpp index 8f112bccb..31750da0d 100644 --- a/src/paimon/core/core_options_test.cpp +++ b/src/paimon/core/core_options_test.cpp @@ -107,6 +107,8 @@ TEST(CoreOptionsTest, TestDefaultValue) { ASSERT_EQ(10, core_options.GetCommitMinRetryWait()); ASSERT_EQ(10 * 1000, core_options.GetCommitMaxRetryWait()); ASSERT_FALSE(core_options.CommitDiscardDuplicateFiles()); + ASSERT_FALSE(core_options.RestCommitLockEnabled()); + ASSERT_FALSE(core_options.RestCommitAcquireLock()); ExpireConfig expire_config = core_options.GetExpireConfig(); ASSERT_EQ(10, expire_config.GetSnapshotRetainMin()); ASSERT_EQ(std::numeric_limits::max(), expire_config.GetSnapshotRetainMax()); @@ -244,6 +246,8 @@ TEST(CoreOptionsTest, TestFromMap) { {Options::COMMIT_MIN_RETRY_WAIT, "5ms"}, {Options::COMMIT_MAX_RETRY_WAIT, "3s"}, {Options::COMMIT_DISCARD_DUPLICATE_FILES, "true"}, + {Options::REST_COMMIT_LOCK_ENABLED, "true"}, + {Options::REST_COMMIT_ACQUIRE_LOCK, "true"}, {Options::DYNAMIC_PARTITION_OVERWRITE, "false"}, {Options::OVERWRITE_UPGRADE, "false"}, {Options::SCAN_SNAPSHOT_ID, "5"}, @@ -398,6 +402,8 @@ TEST(CoreOptionsTest, TestFromMap) { ASSERT_EQ(5, core_options.GetCommitMinRetryWait()); ASSERT_EQ(3 * 1000, core_options.GetCommitMaxRetryWait()); ASSERT_TRUE(core_options.CommitDiscardDuplicateFiles()); + ASSERT_TRUE(core_options.RestCommitLockEnabled()); + ASSERT_TRUE(core_options.RestCommitAcquireLock()); ASSERT_EQ(5, core_options.GetScanSnapshotId().value_or(-1)); ASSERT_EQ(7, core_options.GetScanManifestEntryCacheMaxSnapshots()); ASSERT_FALSE(core_options.ScanManifestEntryLazyDecodeEnabled()); diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp b/src/paimon/core/operation/file_store_commit_impl.cpp index 4d6422c04..dc475dc1c 100644 --- a/src/paimon/core/operation/file_store_commit_impl.cpp +++ b/src/paimon/core/operation/file_store_commit_impl.cpp @@ -290,6 +290,8 @@ Result> FileStoreCommitImpl::ReadAddManifestEntries( } Result FileStoreCommitImpl::RollbackToAsLatest(int64_t target_snapshot_id) { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr lock_scope, + snapshot_commit_->AcquireLock()); PAIMON_ASSIGN_OR_RAISE(std::optional latest_opt, snapshot_manager_->LatestSnapshot()); if (!latest_opt) { return Status::Invalid("Latest snapshot is null, can not roll back."); @@ -330,13 +332,24 @@ Result FileStoreCommitImpl::RollbackToAsLatest(int64_t target_snapshot_id) } } + std::vector new_manifests; std::pair base_manifest_list; std::pair delta_manifest_list; + // Removes the files written below if the rollback stops before sending its snapshot, as when + // its lock is lost. Once sent, they are kept even if uncertain or refused, matching Java. + ScopeGuard unsent_cleanup([&]() { + CleanUpTmpManifests(base_manifest_list.first, delta_manifest_list.first, + /*changelog_list_name=*/std::string(), /*old_metas=*/{}, new_manifests, + /*old_index_manifest=*/std::nullopt, + /*new_index_manifest=*/std::nullopt); + }); PAIMON_ASSIGN_OR_RAISE(std::vector base_manifests, manifest_file_->Write(latest_entries)); + new_manifests.insert(new_manifests.end(), base_manifests.begin(), base_manifests.end()); PAIMON_ASSIGN_OR_RAISE(base_manifest_list, manifest_list_->Write(base_manifests)); PAIMON_ASSIGN_OR_RAISE(std::vector delta_manifests, manifest_file_->Write(delta_files)); + new_manifests.insert(new_manifests.end(), delta_manifests.begin(), delta_manifests.end()); PAIMON_ASSIGN_OR_RAISE(delta_manifest_list, manifest_list_->Write(delta_manifests)); // For row-tracking tables nextRowId must stay monotonic: a rollback to an older snapshot must @@ -366,8 +379,23 @@ Result FileStoreCommitImpl::RollbackToAsLatest(int64_t target_snapshot_id) delta_statistics.push_back(partition_entry); } - PAIMON_ASSIGN_OR_RAISE(bool success, - CommitSnapshotImpl(latest_opt, new_snapshot, delta_statistics)); + Status lock_failure; + Result commit_result = CommitSnapshotImpl(latest_opt, new_snapshot, delta_statistics, + lock_scope.get(), &lock_failure); + if (!commit_result.ok()) { + // The snapshot may have been published before the response was lost. + unsent_cleanup.Release(); + return commit_result.status(); + } + if (!lock_failure.ok()) { + // Unlike a commit, a rollback is not retried once its lock is lost. + return lock_failure.WithMessage(fmt::format( + "Rollback of table {}{} to snapshot {} failed, the commit lock was lost before " + "publishing: {}", + root_path_, CommitTargetSuffix(), target_snapshot_id, lock_failure.message())); + } + unsent_cleanup.Release(); + bool success = commit_result.value(); if (success) { PAIMON_LOG_INFO(logger_, "%s table %s%s to snapshot %ld as new snapshot %ld " @@ -1206,6 +1234,10 @@ Result FileStoreCommitImpl::TryCommit( int32_t retry_count = 0; int64_t start_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000; while (true) { + // Held from reading the head until publication, so the base the attempt builds on is + // covered too. + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr lock_scope, + AcquireLockScope(start_millis, &retry_count)); PAIMON_ASSIGN_OR_RAISE(SnapshotManager::LatestSnapshotResult latest, snapshot_manager_->LatestSnapshotWithSource()); const std::optional& latest_snapshot = latest.snapshot; @@ -1234,14 +1266,18 @@ Result FileStoreCommitImpl::TryCommit( // From publication onward an error may have an uncertain outcome. Keep generated DVs // unless the publisher explicitly reports that this attempt lost the snapshot race. unpublished_cleanup.Release(); + Status lock_failure; PAIMON_ASSIGN_OR_RAISE( bool commit_success, TryCommitOnce(commit_changes->delta_files, commit_changes->changelog_files, commit_changes->index_entries, identifier, watermark, snapshot_properties, - commit_kind, latest_snapshot, detect_conflicts)); + commit_kind, latest_snapshot, detect_conflicts, lock_scope.get(), + &lock_failure)); if (commit_success) { break; } + // Not held while waiting to retry, which would keep other writers out for nothing. + lock_scope.reset(); for (const std::string& temporary_file : commit_changes->temporary_files) { [[maybe_unused]] Status cleanup_status = fs_->Delete(temporary_file, /*recursive=*/false); @@ -1249,29 +1285,50 @@ Result FileStoreCommitImpl::TryCommit( if (!realtime_ranges.empty()) { auto offsets_iter = snapshot_properties.find(RealtimeCommitProperties::kOffsetsKey); if (offsets_iter != snapshot_properties.end()) { - // AtomicStore reported a known conflict, so this attempt's offsets file is not - // referenced by a snapshot and can be removed before rebasing the next attempt. + // AtomicStore reported a known conflict, or the lock stopped the attempt before + // publication, so this attempt's offsets file is not referenced by a snapshot and + // can be removed before rebasing the next attempt. [[maybe_unused]] Status cleanup_status = fs_->Delete(offsets_iter->second, /*recursive=*/false); } } if (!retry_on_conflict) { + if (!lock_failure.ok()) { + return lock_failure.WithMessage(fmt::format( + "real-time commit failed, the commit lock was lost before publishing: {}", + lock_failure.message())); + } return Status::Invalid("real-time commit failed due to snapshot conflict"); } int64_t current_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000; if (current_millis - start_millis > options_.GetCommitTimeout() || retry_count >= options_.GetCommitMaxRetries()) { + if (!lock_failure.ok()) { + return lock_failure.WithMessage(fmt::format( + "Commit to table {}{} failed after {} millis (commit.timeout is {} millis) " + "with {} retries, the commit lock was lost before publishing: {}", + root_path_, CommitTargetSuffix(), current_millis - start_millis, + options_.GetCommitTimeout(), retry_count, lock_failure.message())); + } return Status::Invalid(fmt::format( "Commit to table {}{} failed after {} millis (commit.timeout is {} millis) with " "{} retries, there maybe exist commit conflicts between multiple jobs.", root_path_, CommitTargetSuffix(), current_millis - start_millis, options_.GetCommitTimeout(), retry_count)); } - PAIMON_LOG_INFO(logger_, - "Commit to table %s%s lost a race with a concurrent commit, rebasing and " - "retrying (retry %d of %d).", - root_path_.c_str(), CommitTargetSuffix().c_str(), retry_count + 1, - options_.GetCommitMaxRetries()); + if (!lock_failure.ok()) { + PAIMON_LOG_INFO(logger_, + "Commit to table %s%s lost its commit lock before publishing, " + "retrying under a new one (retry %d of %d).", + root_path_.c_str(), CommitTargetSuffix().c_str(), retry_count + 1, + options_.GetCommitMaxRetries()); + } else { + PAIMON_LOG_INFO(logger_, + "Commit to table %s%s lost a race with a concurrent commit, rebasing " + "and retrying (retry %d of %d).", + root_path_.c_str(), CommitTargetSuffix().c_str(), retry_count + 1, + options_.GetCommitMaxRetries()); + } retry_waiter_.RetryWait(retry_count); retry_count++; } @@ -1344,13 +1401,38 @@ Result> FileStoreCommitImpl::MaxSequenceNumber( return std::optional(max_from_manifest); } +Result> FileStoreCommitImpl::AcquireLockScope( + int64_t start_millis, int32_t* retry_count) { + while (true) { + Result> lock_scope = snapshot_commit_->AcquireLock(); + if (lock_scope.ok()) { + return lock_scope; + } + const Status& failure = lock_scope.status(); + PAIMON_LOG_WARN(logger_, "Failed to acquire commit lock for table %s%s: %s", + root_path_.c_str(), CommitTargetSuffix().c_str(), + failure.ToString().c_str()); + int64_t current_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000; + if (current_millis - start_millis > options_.GetCommitTimeout() || + *retry_count >= options_.GetCommitMaxRetries()) { + return failure.WithMessage(fmt::format( + "Commit to table {}{} failed after {} millis (commit.timeout is {} millis) with " + "{} retries, could not acquire commit lock: {}", + root_path_, CommitTargetSuffix(), current_millis - start_millis, + options_.GetCommitTimeout(), *retry_count, failure.message())); + } + retry_waiter_.RetryWait(*retry_count); + ++(*retry_count); + } +} + Result FileStoreCommitImpl::TryCommitOnce( const std::vector& delta_entries, const std::vector& changelog_entries, const std::vector& index_entries, int64_t identifier, std::optional watermark, const std::map& properties, Snapshot::CommitKind commit_kind, const std::optional& latest_snapshot, - bool detect_conflicts) { + bool detect_conflicts, CommitLockScope* lock_scope, Status* lock_failure) { std::vector delta_files = delta_entries; int64_t start_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000; @@ -1435,6 +1517,7 @@ Result FileStoreCommitImpl::TryCommitOnce( std::vector merge_after_manifests; std::pair base_manifest_list; std::pair delta_manifest_list; + std::optional> changelog_manifest_list; std::vector delta_statistics; std::optional old_index_manifest; @@ -1450,9 +1533,10 @@ Result FileStoreCommitImpl::TryCommitOnce( Snapshot::CommitKind::ToString(commit_kind).c_str(), FormatBaseSnapshotUuid(latest_snapshot).c_str(), commit_time); - CleanUpTmpManifests(base_manifest_list.first, delta_manifest_list.first, - merge_before_manifests, merge_after_manifests, old_index_manifest, - index_manifest_name); + CleanUpTmpManifests( + base_manifest_list.first, delta_manifest_list.first, + changelog_manifest_list ? changelog_manifest_list.value().first : std::string(), + merge_before_manifests, merge_after_manifests, old_index_manifest, index_manifest_name); }); int64_t next_row_id_start = first_row_id_start; int64_t previous_total_record_count = 0; @@ -1526,10 +1610,11 @@ Result FileStoreCommitImpl::TryCommitOnce( PAIMON_ASSIGN_OR_RAISE(delta_manifest_list, manifest_list_->Write(new_changes_manifests)); // write changelog into manifest files - std::optional> changelog_manifest_list; if (!changelog_entries.empty()) { PAIMON_ASSIGN_OR_RAISE(std::vector changelog_manifests, manifest_file_->Write(changelog_entries)); + merge_after_manifests.insert(merge_after_manifests.end(), changelog_manifests.begin(), + changelog_manifests.end()); PAIMON_ASSIGN_OR_RAISE(changelog_manifest_list, manifest_list_->Write(changelog_manifests)); } @@ -1589,8 +1674,8 @@ Result FileStoreCommitImpl::TryCommitOnce( : std::optional>(snapshot_properties), next_row_id_start, new_snapshot_uuid); - Result commit_result = - CommitSnapshotImpl(latest_snapshot, new_snapshot, delta_statistics); + Result commit_result = CommitSnapshotImpl(latest_snapshot, new_snapshot, delta_statistics, + lock_scope, lock_failure); if (!commit_result.ok()) { // The snapshot may have been published before the response was lost. Keep its metadata // and let FilterAndCommit resolve the outcome before retrying. @@ -1668,7 +1753,8 @@ std::string FileStoreCommitImpl::FormatBaseSnapshotUuid(const std::optional FileStoreCommitImpl::CommitSnapshotImpl( const std::optional& base_snapshot, const Snapshot& new_snapshot, - const std::vector& delta_statistics) { + const std::vector& delta_statistics, CommitLockScope* lock_scope, + Status* lock_failure) { std::vector statistics; statistics.reserve(delta_statistics.size()); for (const auto& entry : delta_statistics) { @@ -1678,6 +1764,19 @@ Result FileStoreCommitImpl::CommitSnapshotImpl( } std::optional base_snapshot_uuid = base_snapshot ? base_snapshot.value().Uuid() : std::nullopt; + if (lock_scope != nullptr) { + // Checked last, so a lock lost while the snapshot was prepared stops its publication. + Status lock_status = lock_scope->EnsureValid(); + if (!lock_status.ok()) { + PAIMON_LOG_WARN(logger_, + "Commit lock for table %s%s is no longer valid, not publishing " + "snapshot #%ld: %s", + root_path_.c_str(), CommitTargetSuffix().c_str(), new_snapshot.Id(), + lock_status.ToString().c_str()); + *lock_failure = std::move(lock_status); + return false; + } + } Result commit_result = snapshot_commit_->Commit(base_snapshot_uuid, new_snapshot, snapshot_manager_->Branch(), statistics); if (!commit_result.ok()) { @@ -1696,6 +1795,7 @@ Result FileStoreCommitImpl::CommitSnapshotImpl( void FileStoreCommitImpl::CleanUpTmpManifests( const std::string& base_manifest_list_name, const std::string& delta_manifest_list_name, + const std::string& changelog_manifest_list_name, const std::vector& merge_before_manifests, const std::vector& merge_after_manifests, const std::optional& old_index_manifest, @@ -1708,6 +1808,11 @@ void FileStoreCommitImpl::CleanUpTmpManifests( manifest_list_->DeleteQuietly(delta_manifest_list_name); PAIMON_LOG_DEBUG(logger_, "delta manifest list %s", delta_manifest_list_name.c_str()); } + if (!changelog_manifest_list_name.empty()) { + manifest_list_->DeleteQuietly(changelog_manifest_list_name); + PAIMON_LOG_DEBUG(logger_, "changelog manifest list %s", + changelog_manifest_list_name.c_str()); + } // for faster searching std::set merge_before_manifest_set; for (const auto& merge_before_manifest : merge_before_manifests) { diff --git a/src/paimon/core/operation/file_store_commit_impl.h b/src/paimon/core/operation/file_store_commit_impl.h index 65d6c78e4..ac4b82a2b 100644 --- a/src/paimon/core/operation/file_store_commit_impl.h +++ b/src/paimon/core/operation/file_store_commit_impl.h @@ -28,6 +28,7 @@ #include #include "paimon/common/options/memory_size.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/catalog/snapshot_commit.h" #include "paimon/core/core_options.h" #include "paimon/core/manifest/partition_entry.h" @@ -229,6 +230,16 @@ class FileStoreCommitImpl : public FileStoreCommit { const std::vector>& removed_realtime_partitions, bool detect_conflicts, bool retry_on_conflict); + /// Acquires the lock scope of the next publication attempt. A failed acquisition has prepared + /// and published nothing, so it is retried within the same `commit.max-retries` and + /// `commit.timeout` budget as a conflict, counted in `retry_count`. + Result> AcquireLockScope(int64_t start_millis, + int32_t* retry_count); + + /// @param lock_scope Scope this attempt holds, checked right before publication; null when + /// publication takes no lock. An invalid scope fails the attempt like a conflict. + /// @param lock_failure Set to why `lock_scope` was found invalid, when that is why the + /// attempt returned false. Result TryCommitOnce(const std::vector& delta_files, const std::vector& changelog_files, const std::vector& index_entries, @@ -236,7 +247,8 @@ class FileStoreCommitImpl : public FileStoreCommit { const std::map& properties, Snapshot::CommitKind commit_kind, const std::optional& latest_snapshot, - bool detect_conflicts); + bool detect_conflicts, CommitLockScope* lock_scope, + Status* lock_failure); /// Requires schemas for files carrying row IDs to be available in the table directory. Status CheckRowIdSchemasArePublished( @@ -246,14 +258,21 @@ class FileStoreCommitImpl : public FileStoreCommit { static std::string FormatBaseSnapshotUuid(const std::optional& snapshot); + /// @param lock_scope Scope held for this publication, checked right before + /// `SnapshotCommit::Commit()`; null when publication takes no lock. + /// @param lock_failure Set to why `lock_scope` can no longer protect the publication, in which + /// case nothing is sent and false is returned. + /// @return Whether the snapshot was published; an error leaves the outcome uncertain. Result CommitSnapshotImpl(const std::optional& base_snapshot, const Snapshot& new_snapshot, - const std::vector& delta_statistics); + const std::vector& delta_statistics, + CommitLockScope* lock_scope, Status* lock_failure); Result> ReadAddManifestEntries(const Snapshot& snapshot) const; void CleanUpTmpManifests(const std::string& previous_changes_list_name, const std::string& new_changes_list_name, + const std::string& changelog_list_name, const std::vector& old_metas, const std::vector& new_metas, const std::optional& old_index_manifest, diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp b/src/paimon/core/operation/file_store_commit_impl_test.cpp index 7f0e9032e..96c6de34d 100644 --- a/src/paimon/core/operation/file_store_commit_impl_test.cpp +++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp @@ -45,6 +45,7 @@ #include "paimon/common/utils/linked_hash_map.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/scope_guard.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/catalog/commit_table_request.h" #include "paimon/core/core_options.h" #include "paimon/core/deletionvectors/deletion_vectors_index_file.h" @@ -844,6 +845,92 @@ TEST_F(FileStoreCommitImplTest, TestRequestOnlyCommitToBranch) { ASSERT_GT(CountFiles(PathUtil::JoinPath(table_path_, "manifest")), 0u); } +namespace { + +struct CommitLockLog { + struct Event { + std::string name; + size_t loads; + size_t commits; + }; + + void Record(const std::string& name) { + events.push_back(Event{name, catalog->LoadSnapshotCalls(), catalog->CommitCalls().size()}); + } + + std::vector Names() const { + std::vector names; + for (const Event& event : events) { + names.push_back(event.name); + } + return names; + } + + const MockVersionManagedCatalog* catalog = nullptr; + std::vector events; + std::vector identifiers; + std::vector> table_uuids; + std::vector commit_users; + int32_t failed_acquires = 0; + Status acquire_failure = Status::IOError("lock service unavailable"); + int32_t failed_checks = 0; + Status check_failure = Status::Invalid("the commit lease has expired"); +}; + +class FakeCommitLock : public CommitLock { + public: + explicit FakeCommitLock(std::shared_ptr log) : log_(std::move(log)) {} + + Result> Acquire() override { + log_->Record("acquire"); + if (log_->failed_acquires > 0) { + --log_->failed_acquires; + return log_->acquire_failure; + } + return std::unique_ptr(std::make_unique(log_)); + } + + private: + class Scope : public CommitLockScope { + public: + explicit Scope(std::shared_ptr log) : log_(std::move(log)) {} + + ~Scope() override { + log_->Record("release"); + } + + Status EnsureValid() override { + log_->Record("check"); + if (log_->failed_checks > 0) { + --log_->failed_checks; + return log_->check_failure; + } + return Status::OK(); + } + + private: + std::shared_ptr log_; + }; + + std::shared_ptr log_; +}; + +std::shared_ptr InstallCommitLock(MockVersionManagedCatalog* catalog) { + auto log = std::make_shared(); + log->catalog = catalog; + catalog->SetCommitLockFactory( + [log](const Identifier& identifier, const std::optional& table_uuid, + const std::string& commit_user) -> Result> { + log->identifiers.push_back(identifier); + log->table_uuids.push_back(table_uuid); + log->commit_users.push_back(commit_user); + return std::unique_ptr(std::make_unique(log)); + }); + return log; +} + +} // namespace + TEST_F(FileStoreCommitImplTest, TestCatalogCommitToBranchTakesTheBranchSnapshot) { SchemaManager branch_schema_manager(file_system_, table_path_, "dev"); ASSERT_OK(branch_schema_manager.CreateTable(arrow::schema(fields_), /*partition_keys=*/{"f1"}, @@ -866,12 +953,15 @@ TEST_F(FileStoreCommitImplTest, TestCatalogCommitToBranchTakesTheBranchSnapshot) auto catalog = CreateCatalog(); catalog->SetHeldSnapshot(base_snapshot.value()); + std::shared_ptr log = InstallCommitLock(catalog.get()); CatalogCommitSpec spec; spec.catalog = catalog; spec.identifier.emplace("db", "tbl$branch_dev"); spec.table_id = "table-uuid-1"; ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, CreateCatalogCommit(spec)); ASSERT_OK(commit->Commit(CommitMessagesOfRound(2), /*commit_identifier=*/2)); + ASSERT_EQ(std::vector{Identifier("db", "tbl$branch_dev")}, log->identifiers); + ASSERT_EQ((std::vector{"acquire", "check", "release"}), log->Names()); ASSERT_GT(catalog->LoadSnapshotCalls(), 0u); for (const Identifier& asked : catalog->LoadSnapshotIdentifiers()) { @@ -1126,6 +1216,125 @@ TEST_F(FileStoreCommitImplTest, TestCatalogCommitRetriesWhenRefused) { } } +TEST_F(FileStoreCommitImplTest, TestCatalogCommitHoldsALockScopePerAttempt) { + auto catalog = CreateCatalog(false); + std::shared_ptr log = InstallCommitLock(catalog.get()); + CatalogCommitSpec spec; + spec.catalog = catalog; + spec.table_id = "table-uuid-2"; + spec.options = {{Options::COMMIT_MAX_RETRIES, "1"}, {Options::COMMIT_MIN_RETRY_WAIT, "1ms"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, CreateCatalogCommit(spec)); + ASSERT_EQ(1u, log->identifiers.size()); + ASSERT_EQ(Identifier("db", "tbl"), log->identifiers[0]); + ASSERT_EQ(std::optional("table-uuid-2"), log->table_uuids[0]); + ASSERT_EQ("commit_user_1", log->commit_users[0]); + ASSERT_TRUE(log->events.empty()); + + ASSERT_NOK_WITH_MSG(commit->Commit(CommitMessagesOfRound(1)), "commit conflicts"); + ASSERT_EQ( + (std::vector{"acquire", "check", "release", "acquire", "check", "release"}), + log->Names()); + for (size_t attempt = 0; attempt < 2; ++attempt) { + SCOPED_TRACE(attempt); + const CommitLockLog::Event& acquire = log->events[attempt * 3]; + const CommitLockLog::Event& check = log->events[attempt * 3 + 1]; + const CommitLockLog::Event& release = log->events[attempt * 3 + 2]; + ASSERT_GT(check.loads, acquire.loads); + ASSERT_EQ(attempt, acquire.commits); + ASSERT_EQ(attempt, check.commits); + ASSERT_EQ(attempt + 1, release.commits); + } +} + +TEST_F(FileStoreCommitImplTest, TestCatalogCommitRetriesOnceTheLockScopeIsLost) { + auto failing_catalog = CreateCatalog(); + std::shared_ptr failing_log = InstallCommitLock(failing_catalog.get()); + failing_log->failed_checks = 100; + CatalogCommitSpec spec; + spec.catalog = failing_catalog; + spec.options = {{Options::COMMIT_MAX_RETRIES, "1"}, {Options::COMMIT_MIN_RETRY_WAIT, "1ms"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, CreateCatalogCommit(spec)); + std::vector> msgs = CommitMessagesOfRound(1); + auto* first = dynamic_cast(msgs[0].get()); + ASSERT_NE(nullptr, first); + msgs.push_back(std::make_shared( + first->Partition(), first->Bucket(), first->TotalBuckets(), + DataIncrement({}, {}, {CreateAppendDataFileMeta("changelog-file-1", 10)}), + CompactIncrement({}, {}, {}))); + Status lost = commit->Commit(msgs); + ASSERT_TRUE(lost.IsInvalid()) << lost.ToString(); + ASSERT_NOK_WITH_MSG(lost, "1 retries"); + ASSERT_NOK_WITH_MSG(lost, "the commit lock was lost before publishing"); + ASSERT_NOK_WITH_MSG(lost, "the commit lease has expired"); + ASSERT_EQ(std::string::npos, lost.ToString().find("commit conflicts")) << lost.ToString(); + ASSERT_EQ( + (std::vector{"acquire", "check", "release", "acquire", "check", "release"}), + failing_log->Names()); + ASSERT_TRUE(failing_catalog->CommitCalls().empty()); + std::vector manifests; + ASSERT_OK(file_system_->ListDir(PathUtil::JoinPath(table_path_, "manifest"), &manifests)); + ASSERT_TRUE(manifests.empty()); + + auto catalog = CreateCatalog(); + std::shared_ptr log = InstallCommitLock(catalog.get()); + log->failed_checks = 1; + spec.catalog = catalog; + spec.options = {{Options::COMMIT_MIN_RETRY_WAIT, "1ms"}}; + ASSERT_OK_AND_ASSIGN(commit, CreateCatalogCommit(spec)); + ASSERT_OK(commit->Commit(CommitMessagesOfRound(1))); + ASSERT_EQ( + (std::vector{"acquire", "check", "release", "acquire", "check", "release"}), + log->Names()); + ASSERT_EQ(1u, catalog->CommitCalls().size()); +} + +TEST_F(FileStoreCommitImplTest, TestCatalogCommitRetriesAFailedLockAcquisition) { + auto catalog = CreateCatalog(); + std::shared_ptr log = InstallCommitLock(catalog.get()); + log->failed_acquires = 1; + CatalogCommitSpec spec; + spec.catalog = catalog; + spec.options = {{Options::COMMIT_MIN_RETRY_WAIT, "1ms"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, CreateCatalogCommit(spec)); + ASSERT_OK(commit->Commit(CommitMessagesOfRound(1))); + ASSERT_EQ((std::vector{"acquire", "acquire", "check", "release"}), log->Names()); + ASSERT_EQ(log->events[0].loads, log->events[1].loads); + ASSERT_EQ(1u, catalog->CommitCalls().size()); + + auto failing_catalog = CreateCatalog(); + std::shared_ptr failing_log = InstallCommitLock(failing_catalog.get()); + failing_log->failed_acquires = 100; + spec.catalog = failing_catalog; + spec.options = {{Options::COMMIT_MAX_RETRIES, "2"}, {Options::COMMIT_MIN_RETRY_WAIT, "1ms"}}; + ASSERT_OK_AND_ASSIGN(commit, CreateCatalogCommit(spec)); + Status failed = commit->Commit(CommitMessagesOfRound(1)); + ASSERT_TRUE(failed.IsIOError()) << failed.ToString(); + ASSERT_NOK_WITH_MSG(failed, "could not acquire commit lock"); + ASSERT_NOK_WITH_MSG(failed, "lock service unavailable"); + ASSERT_NOK_WITH_MSG(failed, "2 retries"); + ASSERT_EQ(std::vector(3, "acquire"), failing_log->Names()); + ASSERT_TRUE(failing_catalog->CommitCalls().empty()); +} + +TEST_F(FileStoreCommitImplTest, TestCommitWithoutAUsableLockIsNotCreated) { + auto catalog = CreateCatalog(); + catalog->SetCommitLockFactory([](const Identifier&, const std::optional&, + const std::string&) -> Result> { + return Status::Invalid("no commit lease for this table"); + }); + ASSERT_NOK_WITH_MSG(CreateCatalogCommit(catalog), "no commit lease for this table"); + ASSERT_TRUE(catalog->CommitCalls().empty()); + + CommitContextBuilder builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, + builder.AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::REST_COMMIT_ACQUIRE_LOCK, "true") + .UseRESTCatalogCommit(true) + .WithTableId("table-uuid") + .Finish()); + ASSERT_NOK_WITH_MSG(FileStoreCommit::Create(std::move(context)), "cannot hold a commit lease"); +} + TEST_F(FileStoreCommitImplTest, TestUnknownCatalogOutcomeKeepsTheManifests) { auto catalog = CreateCatalog(Status::IOError("catalog unreachable")); CatalogCommitSpec spec; @@ -2137,6 +2346,38 @@ TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatestThroughCatalog) { ASSERT_FALSE(exist); } +TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatestHoldsALockScope) { + CommitContextBuilder base_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr base_context, + base_builder.AddOption(Options::FILE_SYSTEM, "local").Finish()); + ASSERT_OK_AND_ASSIGN(auto base_commit, FileStoreCommit::Create(std::move(base_context))); + for (int32_t round = 1; round <= 2; ++round) { + ASSERT_OK(base_commit->Commit(CommitMessagesOfRound(round), round)); + } + + auto catalog = CreateCatalog(); + std::shared_ptr log = InstallCommitLock(catalog.get()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, CreateCatalogCommit(catalog)); + ASSERT_OK_AND_ASSIGN(bool rolled_back, commit->RollbackToAsLatest(1)); + ASSERT_TRUE(rolled_back); + ASSERT_EQ((std::vector{"acquire", "check", "release"}), log->Names()); + ASSERT_GT(log->events[1].loads, log->events[0].loads); + ASSERT_EQ(0u, log->events[1].commits); + ASSERT_EQ(1u, log->events[2].commits); + + const std::string manifest_dir = PathUtil::JoinPath(table_path_, "manifest"); + const size_t manifest_count = CountFiles(manifest_dir); + log->failed_checks = 1; + Status lost = commit->RollbackToAsLatest(1).status(); + ASSERT_NOK_WITH_MSG(lost, "the commit lock was lost before publishing"); + ASSERT_NOK_WITH_MSG(lost, "the commit lease has expired"); + ASSERT_EQ(manifest_count, CountFiles(manifest_dir)); + log->failed_acquires = 1; + ASSERT_NOK_WITH_MSG(commit->RollbackToAsLatest(1), "lock service unavailable"); + ASSERT_EQ(manifest_count, CountFiles(manifest_dir)); + ASSERT_EQ(1u, catalog->CommitCalls().size()); +} + TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatestNeedsTheTargetUnderTheTablePath) { CommitContextBuilder base_builder(table_path_, "commit_user_1"); ASSERT_OK_AND_ASSIGN(std::unique_ptr base_context, @@ -2506,10 +2747,10 @@ TEST_F(FileStoreCommitImplTest, TestCleanUpTmpManifests) { table_path_, "manifest/" + index_manifest.value()))); ASSERT_TRUE(exist); - commit_impl->CleanUpTmpManifests(snapshot.value().BaseManifestList(), - snapshot.value().DeltaManifestList(), previous_manifests, - previous_manifests, /*old_index_manifest=*/std::nullopt, - /*new_index_manifest=*/index_manifest); + commit_impl->CleanUpTmpManifests( + snapshot.value().BaseManifestList(), snapshot.value().DeltaManifestList(), + /*changelog_list_name=*/std::string(), previous_manifests, previous_manifests, + /*old_index_manifest=*/std::nullopt, /*new_index_manifest=*/index_manifest); for (const auto& manifest : previous_manifests) { ASSERT_OK_AND_ASSIGN(exist, file_system_->Exists(PathUtil::JoinPath( table_path_, "manifest/" + manifest.FileName()))); @@ -2527,7 +2768,8 @@ TEST_F(FileStoreCommitImplTest, TestCleanUpTmpManifests) { table_path_, "manifest/" + index_manifest.value()))); ASSERT_FALSE(exist); commit_impl->CleanUpTmpManifests( - snapshot.value().BaseManifestList(), snapshot.value().DeltaManifestList(), /*old_metas=*/{}, + snapshot.value().BaseManifestList(), snapshot.value().DeltaManifestList(), + /*changelog_list_name=*/std::string(), /*old_metas=*/{}, /*new_metas=*/previous_manifests, /*old_index_manifest=*/std::nullopt, /*new_index_manifest=*/std::nullopt); for (const auto& manifest : previous_manifests) { diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp index 05751f92e..dd7e18f20 100644 --- a/src/paimon/core/table/table.cpp +++ b/src/paimon/core/table/table.cpp @@ -41,6 +41,7 @@ #include "paimon/common/utils/string_utils.h" #include "paimon/core/catalog/catalog_snapshot_commit.h" #include "paimon/core/catalog/catalog_utils.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/catalog/renaming_snapshot_commit.h" #include "paimon/core/catalog/snapshot_commit.h" #include "paimon/core/catalog/version_managed_catalog.h" @@ -1181,17 +1182,26 @@ bool IsCatalogCommit(const CommitContext& ctx) { Result> NewSnapshotCommit( const CommitContext& ctx, const std::optional& catalog_identifier, - const std::shared_ptr& fs, - const std::shared_ptr& snapshot_manager) { + const CoreOptions& options, const std::shared_ptr& snapshot_manager) { if (IsCatalogCommit(ctx)) { if (!catalog_identifier) { return Status::Invalid("a catalog commit requires a table identifier"); } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr lock, + AsVersionManaged(ctx.GetCatalog()) + ->CreateCommitLock(catalog_identifier.value(), ctx.GetTableId(), + ctx.GetCommitUser(), options)); // Reuse the caller's UUID so table recreation cannot redirect this commit. return std::shared_ptr(std::make_shared( - ctx.GetCatalog(), catalog_identifier.value(), ctx.GetTableId())); + ctx.GetCatalog(), catalog_identifier.value(), ctx.GetTableId(), std::move(lock))); } if (ctx.UseRESTCatalogCommit()) { + if (options.RestCommitAcquireLock()) { + return Status::Invalid(fmt::format( + "'{}' needs a commit through a REST catalog with WithCatalog(); a commit table " + "request built for the caller to send cannot hold a commit lease", + Options::REST_COMMIT_ACQUIRE_LOCK)); + } return std::shared_ptr( std::make_shared(ctx.GetTableId())); } @@ -1203,7 +1213,7 @@ Result> NewSnapshotCommit( "catalog with WithCatalog() or ask for the request with UseRESTCatalogCommit()"); } return std::shared_ptr( - std::make_shared(fs, snapshot_manager)); + std::make_shared(options.GetFileSystem(), snapshot_manager)); } /// Creates a format-table commit, rejecting non-default options that require snapshots. @@ -1417,9 +1427,8 @@ Result> FileStoreCommit::Create( ctx->GetExecutor(), ctx->GetMemoryPool()); } - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr snapshot_commit, - NewSnapshotCommit(*ctx, catalog_identifier, options.GetFileSystem(), snapshot_manager)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot_commit, + NewSnapshotCommit(*ctx, catalog_identifier, options, snapshot_manager)); return std::make_unique( ctx->GetMemoryPool(), ctx->GetExecutor(), arrow_schema, root_path, ctx->GetCommitUser(), diff --git a/src/paimon/rest/resource_paths.cpp b/src/paimon/rest/resource_paths.cpp index e1ee40f6c..75cfad26e 100644 --- a/src/paimon/rest/resource_paths.cpp +++ b/src/paimon/rest/resource_paths.cpp @@ -68,6 +68,16 @@ std::string ResourcePaths::CommitTable(const std::string& database_name, return Table(database_name, table_name) + "/commit"; } +std::string ResourcePaths::CommitLock(const std::string& database_name, + const std::string& table_name) const { + return Table(database_name, table_name) + "/commit-lock"; +} + +std::string ResourcePaths::RenewCommitLock(const std::string& database_name, + const std::string& table_name) const { + return CommitLock(database_name, table_name) + "/renew"; +} + std::string ResourcePaths::TableSnapshot(const std::string& database_name, const std::string& table_name) const { return Table(database_name, table_name) + "/snapshot"; diff --git a/src/paimon/rest/resource_paths.h b/src/paimon/rest/resource_paths.h index 9294beffe..0e7bf5f6f 100644 --- a/src/paimon/rest/resource_paths.h +++ b/src/paimon/rest/resource_paths.h @@ -39,6 +39,11 @@ class ResourcePaths { std::string RenameTable() const; std::string Snapshots(const std::string& database_name, const std::string& table_name) const; std::string CommitTable(const std::string& database_name, const std::string& table_name) const; + /// Endpoint acquiring the commit lease of a table or branch. + std::string CommitLock(const std::string& database_name, const std::string& table_name) const; + /// Endpoint renewing the commit lease of a table or branch. + std::string RenewCommitLock(const std::string& database_name, + const std::string& table_name) const; /// Endpoint for the table's current snapshot. std::string TableSnapshot(const std::string& database_name, const std::string& table_name) const; diff --git a/src/paimon/rest/resource_paths_test.cpp b/src/paimon/rest/resource_paths_test.cpp index bc8a1355b..5a6cd3e77 100644 --- a/src/paimon/rest/resource_paths_test.cpp +++ b/src/paimon/rest/resource_paths_test.cpp @@ -32,6 +32,11 @@ TEST(ResourcePathsTest, WithPrefix) { ASSERT_EQ("/v1/my+prefix/tables/rename", paths.RenameTable()); ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1/snapshots", paths.Snapshots("db", "t1")); ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1/commit", paths.CommitTable("db", "t1")); + ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1/commit-lock", paths.CommitLock("db", "t1")); + ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1/commit-lock/renew", + paths.RenewCommitLock("db", "t1")); + ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1%24branch_b1/commit-lock", + paths.CommitLock("db", "t1$branch_b1")); ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1/snapshot", paths.TableSnapshot("db", "t1")); ASSERT_EQ("/v1/my+prefix/databases/db/tables/t1/token", paths.TableToken("db", "t1")); ASSERT_EQ("/v1/my+prefix/databases/db%231/tables/t+1/token", paths.TableToken("db#1", "t 1")); diff --git a/src/paimon/rest/rest_api.cpp b/src/paimon/rest/rest_api.cpp index 45865595d..5862fb355 100644 --- a/src/paimon/rest/rest_api.cpp +++ b/src/paimon/rest/rest_api.cpp @@ -202,7 +202,7 @@ Status RestApi::ErrorToStatus(const RestHttpClient::Response& response, bool fol Result RestApi::Execute( const std::string& method, const std::string& path, const std::map& query_params, const std::string& body, - bool retry_safe) const { + bool retry_safe, const RestHttpClient::Cancellation* cancellation) const { RestAuthParameter auth_parameter = RestAuthParameter::Create(method, path, query_params, body); StringMap request_headers = base_headers_; if (!body.empty()) { @@ -212,9 +212,9 @@ Result RestApi::Execute( PAIMON_ASSIGN_OR_RAISE(StringMap headers, auth_provider_->MergeAuthHeader(request_headers, auth_parameter)); bool follow_redirects = auth_provider_->AllowsRedirects(); - PAIMON_ASSIGN_OR_RAISE( - RestHttpClient::Response response, - client_->Execute(method, path, query_params, headers, body, follow_redirects, retry_safe)); + PAIMON_ASSIGN_OR_RAISE(RestHttpClient::Response response, + client_->Execute(method, path, query_params, headers, body, + follow_redirects, retry_safe, cancellation)); if (!response.IsSuccessful()) { return ErrorToStatus(response, follow_redirects); } @@ -339,6 +339,35 @@ Result RestApi::CommitSnapshot(const Identifier& identifier, return entity.IsSuccess(); } +Result RestApi::AcquireCommitLock(const Identifier& identifier, + const std::string& table_uuid, + const std::string& commit_user) const { + return PostCommitLock( + resource_paths_.CommitLock(identifier.GetDatabaseName(), identifier.GetObjectName()), + table_uuid, commit_user, /*cancellation=*/nullptr); +} + +Result RestApi::RenewCommitLock( + const Identifier& identifier, const std::string& table_uuid, const std::string& commit_user, + const RestHttpClient::Cancellation* cancellation) const { + return PostCommitLock( + resource_paths_.RenewCommitLock(identifier.GetDatabaseName(), identifier.GetObjectName()), + table_uuid, commit_user, cancellation); +} + +Result RestApi::PostCommitLock( + const std::string& path, const std::string& table_uuid, const std::string& commit_user, + const RestHttpClient::Cancellation* cancellation) const { + CommitLockRequest request(table_uuid, commit_user); + PAIMON_ASSIGN_OR_RAISE(std::string body, request.ToJsonString()); + // Safe to replay: the same commit user shares a lease, and a renewal never revives one. + PAIMON_ASSIGN_OR_RAISE(RestHttpClient::Response response, + Execute("POST", path, {}, body, /*retry_safe=*/true, cancellation)); + CommitLockResponse entity; + PAIMON_RETURN_NOT_OK(ParseResponseBody(response.body, path, &entity)); + return entity; +} + Result RestApi::LoadTableToken(const Identifier& identifier) const { return GetEntity( resource_paths_.TableToken(identifier.GetDatabaseName(), identifier.GetObjectName()), {}); diff --git a/src/paimon/rest/rest_api.h b/src/paimon/rest/rest_api.h index 722fc525c..d42095501 100644 --- a/src/paimon/rest/rest_api.h +++ b/src/paimon/rest/rest_api.h @@ -133,6 +133,28 @@ class RestApi { const Snapshot& snapshot, const std::vector& statistics) const; + /// Acquires the commit lease of a table or branch for `commit_user`. Contention is reported + /// as `acquired=false`, not as an error. + /// + /// @param identifier Table or branch the lease guards. + /// @param table_uuid Catalog UUID of the table or branch. + /// @param commit_user Exact commit user of the snapshots published under the lease. + Result AcquireCommitLock(const Identifier& identifier, + const std::string& table_uuid, + const std::string& commit_user) const; + + /// Renews the lease `commit_user` holds. A renewal never acquires a new lease, so + /// `acquired=false` means the lease was lost or has expired. + /// + /// @param identifier Table or branch the lease guards. + /// @param table_uuid Catalog UUID of the table or branch. + /// @param commit_user Exact commit user the lease was granted to. + /// @param cancellation Ends the renewal with a `Cancelled` status once cancelled, including + /// while it waits for the response or to retry; null to wait for the outcome. + Result RenewCommitLock( + const Identifier& identifier, const std::string& table_uuid, const std::string& commit_user, + const RestHttpClient::Cancellation* cancellation = nullptr) const; + /// Loads the temporary file system credentials of one table. A 403 means the caller /// has no permission for the table and is reported as an error status carrying a /// `RestErrorDetail`. @@ -152,15 +174,22 @@ class RestApi { const std::map& options, const ResourcePaths& paths); /// Executes an authenticated request and maps unsuccessful responses to errors. - /// `retry_safe=false` disables retries and redirects. - Result Execute(const std::string& method, const std::string& path, - const std::map& query_params, - const std::string& body, bool retry_safe = true) const; + /// `retry_safe=false` disables retries and redirects. See `RestHttpClient::Execute()` for + /// `cancellation`. + Result Execute( + const std::string& method, const std::string& path, + const std::map& query_params, const std::string& body, + bool retry_safe = true, const RestHttpClient::Cancellation* cancellation = nullptr) const; template Result GetEntity(const std::string& path, const std::map& query_params) const; + /// Posts a commit lease request to `path`, which acquiring and renewing share. + Result PostCommitLock( + const std::string& path, const std::string& table_uuid, const std::string& commit_user, + const RestHttpClient::Cancellation* cancellation) const; + /// Fetches all pages, stopping at an empty page or a missing/empty next page token. template Result> ListAllPages(const std::string& path) const; diff --git a/src/paimon/rest/rest_catalog.cpp b/src/paimon/rest/rest_catalog.cpp index 66898d8af..273804d0d 100644 --- a/src/paimon/rest/rest_catalog.cpp +++ b/src/paimon/rest/rest_catalog.cpp @@ -39,6 +39,7 @@ #include "paimon/core/table/system/system_table_schema.h" #include "paimon/defs.h" #include "paimon/fs/file_system.h" +#include "paimon/rest/rest_commit_lock.h" #include "paimon/rest/rest_credential_provider.h" #include "paimon/rest/rest_token_file_system.h" #include "paimon/rest/rest_util.h" @@ -547,4 +548,23 @@ Result RestCatalog::CommitSnapshot(const Identifier& identifier, statistics); } +Result> RestCatalog::CreateCommitLock( + const Identifier& identifier, const std::optional& table_uuid, + const std::string& commit_user, const CoreOptions& options) { + if (!options.RestCommitAcquireLock()) { + return std::unique_ptr(); + } + if (!options.RestCommitLockEnabled()) { + return Status::Invalid(fmt::format("{} requires {} on the table.", + Options::REST_COMMIT_ACQUIRE_LOCK, + Options::REST_COMMIT_LOCK_ENABLED)); + } + PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotSystemTable(identifier, "createCommitLock")); + PAIMON_ASSIGN_OR_RAISE(Identifier load_identifier, ToLoadIdentifier(identifier)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr lock, + RestCommitLock::Create(api_, load_identifier, table_uuid, commit_user, + options.GetCommitTimeout())); + return std::unique_ptr(std::move(lock)); +} + } // namespace paimon diff --git a/src/paimon/rest/rest_catalog.h b/src/paimon/rest/rest_catalog.h index 051b90d38..2f57e912d 100644 --- a/src/paimon/rest/rest_catalog.h +++ b/src/paimon/rest/rest_catalog.h @@ -107,6 +107,13 @@ class RestCatalog : public Catalog, public VersionManagedCatalog { const Snapshot& snapshot, const std::vector& statistics) override; + /// Returns a `RestCommitLock` when the writer sets `Options::REST_COMMIT_ACQUIRE_LOCK`, which + /// requires `Options::REST_COMMIT_LOCK_ENABLED` on the table, and null otherwise. No request + /// is sent until a lease is acquired. + Result> CreateCommitLock( + const Identifier& identifier, const std::optional& table_uuid, + const std::string& commit_user, const CoreOptions& options) override; + /// Options merged with the server side config. const std::map& GetOptions() const override; diff --git a/src/paimon/rest/rest_catalog_test.cpp b/src/paimon/rest/rest_catalog_test.cpp index 4b878250d..147a908ce 100644 --- a/src/paimon/rest/rest_catalog_test.cpp +++ b/src/paimon/rest/rest_catalog_test.cpp @@ -20,18 +20,21 @@ #include #include +#include #include #include #include #include #include #include +#include #include #include #include "arrow/api.h" #include "arrow/c/bridge.h" #include "fmt/format.h" +#include "fmt/ranges.h" #include "gtest/gtest.h" #include "paimon/catalog/catalog.h" #include "paimon/catalog/table.h" @@ -41,7 +44,9 @@ #include "paimon/common/utils/checked_cast.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/string_utils.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/catalog/commit_table_request.h" +#include "paimon/core/core_options.h" #include "paimon/core/partition/partition_statistics.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/utils/snapshot_manager.h" @@ -92,6 +97,15 @@ struct MockCatalogState { std::string last_commit_body; std::string last_commit_table; bool refuse_commit = false; + int32_t refused_commits_left = 0; + std::vector lock_requests; + std::vector lock_tables; + int32_t contended_acquires = 0; + int64_t lease_millis = 60000; + bool lose_on_renew = false; + int64_t snapshot_delay_millis = 0; + int64_t get_table_delay_millis = 0; + std::vector table_events; bool break_commit_response = false; bool commit_response_without_outcome = false; bool accept_commit_then_report_unavailable = false; @@ -331,12 +345,26 @@ MockRestServer::Response HandleCatalogRequest(MockCatalogState* state, bool list_snapshots = false; bool commit_table = false; bool load_snapshot = false; + bool acquire_lock = false; + bool renew_lock = false; const std::string snapshots_suffix = "/snapshots"; const std::string snapshot_suffix = "/snapshot"; const std::string commit_suffix = "/commit"; - if (table_name.size() > snapshots_suffix.size() && - table_name.compare(table_name.size() - snapshots_suffix.size(), snapshots_suffix.size(), - snapshots_suffix) == 0) { + const std::string acquire_lock_suffix = "/commit-lock"; + const std::string renew_lock_suffix = "/commit-lock/renew"; + if (table_name.size() > renew_lock_suffix.size() && + table_name.compare(table_name.size() - renew_lock_suffix.size(), renew_lock_suffix.size(), + renew_lock_suffix) == 0) { + table_name = table_name.substr(0, table_name.size() - renew_lock_suffix.size()); + renew_lock = true; + } else if (table_name.size() > acquire_lock_suffix.size() && + table_name.compare(table_name.size() - acquire_lock_suffix.size(), + acquire_lock_suffix.size(), acquire_lock_suffix) == 0) { + table_name = table_name.substr(0, table_name.size() - acquire_lock_suffix.size()); + acquire_lock = true; + } else 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; } else if (table_name.size() > commit_suffix.size() && @@ -354,12 +382,39 @@ MockRestServer::Response HandleCatalogRequest(MockCatalogState* state, if (table_iter == tables.end()) { return MockError(404, ErrorResponse::kResourceTypeTable, table_name, "table not found"); } + if (acquire_lock || renew_lock) { + if (request.method != "POST") { + return MockError(400, "", "", "unsupported method"); + } + Result lock_request = CommitLockRequest::FromJsonString(request.body); + if (!lock_request.ok()) { + return MockError(400, "", "", "bad commit lock request"); + } + state->lock_requests.push_back(lock_request.value()); + state->lock_tables.push_back(table_name); + state->table_events.push_back(acquire_lock ? "acquire" : "renew"); + bool contended = acquire_lock && state->contended_acquires > 0; + if (contended) { + --state->contended_acquires; + } + if (contended || (renew_lock && state->lose_on_renew)) { + CommitLockResponse not_granted(/*acquired=*/false, /*commit_user=*/std::nullopt, + /*expires_at_millis=*/0, /*lease_millis=*/0, + /*snapshot=*/std::nullopt); + return JsonResponse(200, not_granted.ToJsonString().value()); + } + CommitLockResponse response(/*acquired=*/true, lock_request.value().GetCommitUser(), + /*expires_at_millis=*/1700000000000 + state->lease_millis, + state->lease_millis, /*snapshot=*/std::nullopt); + return JsonResponse(200, response.ToJsonString().value()); + } if (commit_table) { if (request.method != "POST") { return MockError(400, "", "", "unsupported method"); } state->last_commit_body = request.body; state->last_commit_table = table_name; + state->table_events.push_back("commit"); ++state->commit_requests; if (state->accept_commit_then_report_unavailable) { if (state->commit_requests == 1) { @@ -402,7 +457,12 @@ MockRestServer::Response HandleCatalogRequest(MockCatalogState* state, if (state->commit_response_without_outcome) { return JsonResponse(200, R"({"success": null})"); } - CommitTableResponse response(!state->refuse_commit); + bool refuse = state->refuse_commit; + if (state->refused_commits_left > 0) { + --state->refused_commits_left; + refuse = true; + } + CommitTableResponse response(!refuse); return JsonResponse(200, response.ToJsonString().value()); } if (load_snapshot) { @@ -410,6 +470,10 @@ MockRestServer::Response HandleCatalogRequest(MockCatalogState* state, return MockError(400, "", "", "unsupported method"); } state->last_snapshot_table = table_name; + state->table_events.push_back("snapshot"); + if (state->snapshot_delay_millis > 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(state->snapshot_delay_millis)); + } if (state->null_snapshot_response) { return JsonResponse(200, R"({"snapshot": null})"); } @@ -433,6 +497,9 @@ MockRestServer::Response HandleCatalogRequest(MockCatalogState* state, } if (request.method == "GET") { state->get_table_requests++; + if (state->get_table_delay_millis > 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(state->get_table_delay_millis)); + } return JsonResponse(200, TableResponseJson(table_name, table_iter->second)); } if (request.method == "DELETE") { @@ -1915,6 +1982,292 @@ TEST_F(RestCatalogTest, CommitTakenThenReportedUnavailableIsNotReplayed) { } } +TEST_F(RestCatalogTest, CommitLockAddressesTheTableOrBranch) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr catalog, CreateRestCatalog()); + ASSERT_OK(catalog->CreateDatabase("db1", {}, false)); + Identifier identifier("db1", "t1"); + ASSERT_OK(CreateSampleTable(catalog.get(), identifier)); + ASSERT_OK_AND_ASSIGN(CoreOptions options, + CoreOptions::FromMap({{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "mock_format"}, + {Options::REST_COMMIT_LOCK_ENABLED, "true"}, + {Options::REST_COMMIT_ACQUIRE_LOCK, "true"}})); + ASSERT_NOK_WITH_MSG(catalog->CreateCommitLock(Identifier("db1", "t1$snapshots"), "table-uuid", + "commit-user", options), + "Cannot 'createCommitLock' for system table"); + + MockCatalogState::TableData branch_data; + branch_data.schema_json = R"({"fields": [{"id": 0, "name": "f0", "type": "INT NOT NULL"}],)" + R"( "partitionKeys": [], "primaryKeys": [], "options": {}})"; + branch_data.path = "wh1/db1.db/t1"; + { + std::lock_guard guard(state_->mutex); + state_->databases["db1"]["t1$branch_b1"] = branch_data; + } + struct Case { + Identifier identifier; + std::string served_table; + }; + for (const Case& lease_case : + {Case{identifier, "t1"}, Case{Identifier("db1", "t1$branch_b1"), "t1$branch_b1"}, + Case{Identifier("db1", "t1$branch_main"), "t1"}}) { + SCOPED_TRACE(lease_case.identifier.ToString()); + size_t requests_before = 0; + { + std::lock_guard guard(state_->mutex); + requests_before = state_->lock_requests.size(); + } + ASSERT_OK_AND_ASSIGN( + std::unique_ptr lock, + catalog->CreateCommitLock(lease_case.identifier, "table-uuid", "commit-user", options)); + ASSERT_TRUE(lock); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, lock->Acquire()); + ASSERT_OK(scope->EnsureValid()); + std::lock_guard guard(state_->mutex); + ASSERT_EQ(requests_before + 1, state_->lock_requests.size()); + ASSERT_EQ(lease_case.served_table, state_->lock_tables.back()); + ASSERT_EQ("table-uuid", state_->lock_requests.back().GetTableId()); + ASSERT_EQ("commit-user", state_->lock_requests.back().GetCommitUser()); + } +} + +class RestCatalogCommitLeaseTest : public RestCatalogTest { + protected: + void SetUp() override { + RestCatalogTest::SetUp(); + dir_ = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir_); + } + + Result> CreateCatalogServingTable(bool lock_enabled) { + std::string lock_option = lock_enabled ? R"(, "rest.commit.lock-enabled": "true")" : ""; + MockCatalogState::TableData table_data; + table_data.schema_json = R"({"fields": [{"id": 0, "name": "f0", "type": "STRING"},)" + R"( {"id": 1, "name": "f1", "type": "INT"},)" + R"( {"id": 2, "name": "f2", "type": "INT"},)" + R"( {"id": 3, "name": "f3", "type": "DOUBLE"}],)" + R"( "partitionKeys": ["f1"], "primaryKeys": [],)" + R"( "options": {"file.format": "orc", "manifest.format": "avro")" + + lock_option + "}}"; + table_data.path = dir_->Str(); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr catalog, + CreateRestCatalog(FastRetryConfig())); + PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("db1", {}, /*ignore_if_exists=*/true)); + { + std::lock_guard lock(state_->mutex); + state_->databases["db1"]["t1"] = table_data; + } + return std::shared_ptr(std::move(catalog)); + } + + Result> CreateCommit( + const std::shared_ptr& catalog, const std::optional& table_id, + const std::map& options) { + CommitContextBuilder builder(dir_->Str(), "commit-user"); + builder.WithCatalog(catalog, Identifier("db1", "t1")); + if (table_id) { + builder.WithTableId(table_id.value()); + } + for (const auto& [key, value] : options) { + builder.AddOption(key, value); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr context, builder.Finish()); + return FileStoreCommit::Create(std::move(context)); + } + + std::vector TableEvents() { + std::lock_guard lock(state_->mutex); + return state_->table_events; + } + + size_t LockRequests() { + std::lock_guard lock(state_->mutex); + return state_->lock_requests.size(); + } + + void AssertNoManifestsLeft() { + std::shared_ptr fs = dir_->GetFileSystem(); + const std::string manifest_dir = PathUtil::JoinPath(dir_->Str(), "manifest"); + ASSERT_OK_AND_ASSIGN(bool manifest_dir_exists, fs->Exists(manifest_dir)); + if (manifest_dir_exists) { + std::vector manifests; + ASSERT_OK(fs->ListDir(manifest_dir, &manifests)); + ASSERT_TRUE(manifests.empty()) << manifests.size(); + } + } + + std::unique_ptr dir_; +}; + +TEST_F(RestCatalogCommitLeaseTest, LeaseIsAcquiredBeforeTheHeadAndAgainAfterAConflict) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr catalog, + CreateCatalogServingTable(/*lock_enabled=*/true)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + CreateCommit(catalog, "1", + {{Options::REST_COMMIT_ACQUIRE_LOCK, "true"}, + {Options::COMMIT_MIN_RETRY_WAIT, "1ms"}})); + ASSERT_EQ(0u, LockRequests()); + { + std::lock_guard lock(state_->mutex); + state_->refused_commits_left = 1; + state_->table_events.clear(); + } + ASSERT_OK_AND_ASSIGN(std::vector> msgs, + ReadFixtureCommitMessages(dir_->GetFileSystem())); + ASSERT_OK(commit->Commit(msgs, 7)); + + std::vector events = TableEvents(); + std::vector leases_and_commits; + bool leased = false; + bool head_read_under_lease = false; + for (const std::string& event : events) { + if (event == "snapshot") { + head_read_under_lease = head_read_under_lease || leased; + continue; + } + leases_and_commits.push_back(event); + if (event == "acquire") { + leased = true; + head_read_under_lease = false; + } else if (event == "commit") { + ASSERT_TRUE(leased && head_read_under_lease) + << fmt::format("{}", fmt::join(events, ",")); + leased = false; + } + } + ASSERT_EQ((std::vector{"acquire", "commit", "acquire", "commit"}), + leases_and_commits); + + std::lock_guard lock(state_->mutex); + ASSERT_EQ(2u, state_->lock_requests.size()); + for (size_t i = 0; i < state_->lock_requests.size(); ++i) { + ASSERT_EQ("t1", state_->lock_tables[i]); + ASSERT_EQ("1", state_->lock_requests[i].GetTableId()); + ASSERT_EQ("commit-user", state_->lock_requests[i].GetCommitUser()); + } + ASSERT_OK_AND_ASSIGN(CommitTableRequest published, + CommitTableRequest::FromJsonString(state_->last_commit_body)); + ASSERT_EQ(std::optional("1"), published.GetTableId()); + ASSERT_EQ("commit-user", published.GetSnapshot().CommitUser()); +} + +TEST_F(RestCatalogCommitLeaseTest, LeaseNeedsBothOptionsAndTheTableUuid) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr catalog, + CreateCatalogServingTable(/*lock_enabled=*/true)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, CreateCommit(catalog, "1", {})); + ASSERT_OK_AND_ASSIGN(std::vector> msgs, + ReadFixtureCommitMessages(dir_->GetFileSystem())); + ASSERT_OK(commit->Commit(msgs, 7)); + { + std::lock_guard lock(state_->mutex); + ASSERT_EQ(1, state_->commit_requests); + } + ASSERT_NOK_WITH_MSG( + CreateCommit(catalog, std::nullopt, {{Options::REST_COMMIT_ACQUIRE_LOCK, "true"}}), + "requires the table UUID"); + + ASSERT_OK_AND_ASSIGN(catalog, CreateCatalogServingTable(/*lock_enabled=*/false)); + Status refused = + CreateCommit(catalog, "1", {{Options::REST_COMMIT_ACQUIRE_LOCK, "true"}}).status(); + ASSERT_TRUE(refused.IsInvalid()) << refused.ToString(); + ASSERT_NOK_WITH_MSG(refused, + "rest.commit.acquire-lock requires rest.commit.lock-enabled on the table."); + ASSERT_EQ(0u, LockRequests()); +} + +TEST_F(RestCatalogCommitLeaseTest, ContendedLeaseIsAwaitedUntilCommitTimeout) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr catalog, + CreateCatalogServingTable(/*lock_enabled=*/true)); + ASSERT_OK_AND_ASSIGN(std::vector> msgs, + ReadFixtureCommitMessages(dir_->GetFileSystem())); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + CreateCommit(catalog, "1", + {{Options::REST_COMMIT_ACQUIRE_LOCK, "true"}, + {Options::COMMIT_MAX_RETRIES, "0"}})); + { + std::lock_guard lock(state_->mutex); + state_->contended_acquires = 2; + } + ASSERT_OK(commit->Commit(msgs, 7)); + { + std::lock_guard lock(state_->mutex); + ASSERT_EQ(3u, state_->lock_requests.size()); + ASSERT_EQ(1, state_->commit_requests); + } + + ASSERT_OK_AND_ASSIGN(commit, CreateCommit(catalog, "1", + {{Options::REST_COMMIT_ACQUIRE_LOCK, "true"}, + {Options::COMMIT_TIMEOUT, "0ms"}, + {Options::COMMIT_MAX_RETRIES, "0"}})); + { + std::lock_guard lock(state_->mutex); + state_->contended_acquires = 1000; + } + Status failed = commit->Commit(msgs, 8); + ASSERT_NOK_WITH_MSG(failed, "could not acquire commit lock"); + ASSERT_NOK_WITH_MSG(failed, "timed out acquiring the commit lock"); + std::lock_guard lock(state_->mutex); + ASSERT_EQ(4u, state_->lock_requests.size()); + ASSERT_EQ(1, state_->commit_requests); +} + +TEST_F(RestCatalogCommitLeaseTest, LeaseExpiredLocallyBeforePublicationSendsNoCommit) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr catalog, + CreateCatalogServingTable(/*lock_enabled=*/true)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + CreateCommit(catalog, "1", + {{Options::REST_COMMIT_ACQUIRE_LOCK, "true"}, + {Options::COMMIT_MAX_RETRIES, "1"}, + {Options::COMMIT_MIN_RETRY_WAIT, "1ms"}})); + { + std::lock_guard lock(state_->mutex); + state_->lease_millis = 60; + state_->snapshot_delay_millis = 200; + state_->table_events.clear(); + } + ASSERT_OK_AND_ASSIGN(std::vector> msgs, + ReadFixtureCommitMessages(dir_->GetFileSystem())); + Status failed = commit->Commit(msgs, 7); + ASSERT_TRUE(failed.IsInvalid()) << failed.ToString(); + ASSERT_NOK_WITH_MSG(failed, "the commit lock was lost before publishing"); + ASSERT_NOK_WITH_MSG(failed, "is no longer usable"); + ASSERT_NOK_WITH_MSG(failed, "expired locally"); + ASSERT_EQ(std::string::npos, failed.ToString().find("commit conflicts")) << failed.ToString(); + std::vector events = TableEvents(); + ASSERT_EQ(2, std::count(events.begin(), events.end(), std::string("acquire"))); + ASSERT_EQ(0, std::count(events.begin(), events.end(), std::string("commit"))); + AssertNoManifestsLeft(); +} + +TEST_F(RestCatalogCommitLeaseTest, RenewalRefusedBeforePublicationSendsNoCommit) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr catalog, + CreateCatalogServingTable(/*lock_enabled=*/true)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + CreateCommit(catalog, "1", + {{Options::REST_COMMIT_ACQUIRE_LOCK, "true"}, + {Options::COMMIT_MAX_RETRIES, "0"}})); + { + std::lock_guard lock(state_->mutex); + state_->lease_millis = 900; + state_->lose_on_renew = true; + state_->snapshot_delay_millis = 600; + state_->get_table_delay_millis = 100; + state_->table_events.clear(); + } + ASSERT_OK_AND_ASSIGN(std::vector> msgs, + ReadFixtureCommitMessages(dir_->GetFileSystem())); + Status failed = commit->Commit(msgs, 7); + ASSERT_TRUE(failed.IsInvalid()) << failed.ToString(); + ASSERT_NOK_WITH_MSG(failed, "the commit lock was lost before publishing"); + ASSERT_NOK_WITH_MSG(failed, "is no longer usable: the commit lease has expired"); + ASSERT_EQ(std::string::npos, failed.ToString().find("expired locally")) << failed.ToString(); + std::vector events = TableEvents(); + ASSERT_EQ(1, std::count(events.begin(), events.end(), std::string("acquire"))); + ASSERT_EQ(1, std::count(events.begin(), events.end(), std::string("renew"))); + ASSERT_EQ(0, std::count(events.begin(), events.end(), std::string("commit"))); + AssertNoManifestsLeft(); +} + TEST(RestApiErrorTest, ErrorToStatus) { RestHttpClient::Response response; response.code = 404; diff --git a/src/paimon/rest/rest_commit_lock.cpp b/src/paimon/rest/rest_commit_lock.cpp new file mode 100644 index 000000000..f12007d85 --- /dev/null +++ b/src/paimon/rest/rest_commit_lock.cpp @@ -0,0 +1,286 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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_commit_lock.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "fmt/format.h" +#include "paimon/rest/rest_api.h" +#include "paimon/rest/rest_http_client.h" +#include "paimon/rest/rest_messages.h" + +namespace paimon { + +namespace { + +// Longest pause between two acquisition requests while another writer holds the lease. +constexpr int64_t kMaxPollIntervalMillis = 100; + +Status ValidateGrant(const CommitLockResponse& response, const std::string& commit_user) { + const std::optional& owner = response.GetCommitUser(); + if (owner != commit_user) { + return Status::Invalid(fmt::format( + "the server returned an invalid commit lease: expected owner '{}', got {}", commit_user, + owner ? fmt::format("'{}'", owner.value()) : std::string("null"))); + } + if (response.GetLeaseMillis() <= 0) { + return Status::Invalid(fmt::format( + "the server returned an invalid commit lease: expected a positive leaseMillis, got {}", + response.GetLeaseMillis())); + } + return Status::OK(); +} + +// A granted lease, shared by its scope and the thread renewing it, which may outlive the scope. +class LeaseState { + public: + LeaseState(std::shared_ptr api, Identifier identifier, std::string table_uuid, + std::string commit_user, RestCommitLock::Clock clock, int64_t lease_millis, + int64_t requested_at_millis, std::shared_ptr logger) + : api_(std::move(api)), + identifier_(std::move(identifier)), + table_uuid_(std::move(table_uuid)), + commit_user_(std::move(commit_user)), + clock_(std::move(clock)), + lease_millis_(lease_millis), + logger_(std::move(logger)), + renewed_at_millis_(requested_at_millis) {} + + Status EnsureValid() { + std::lock_guard lock(mutex_); + return CheckValidLocked(); + } + + void Close() { + { + std::lock_guard lock(mutex_); + closed_ = true; + } + closed_cv_.notify_all(); + renewal_cancellation_.Cancel(); + } + + void RenewLoop() { + const std::chrono::milliseconds interval(std::max(1, lease_millis_ / 3)); + std::unique_lock lock(mutex_); + while (true) { + const bool closed = closed_cv_.wait_for(lock, interval, [this]() { return closed_; }); + if (closed) { + return; + } + if (!CheckValidLocked().ok()) { + return; + } + const int64_t requested_at = clock_(); + lock.unlock(); + Status renewed = Renew(); + lock.lock(); + // A late answer cannot revive a scope that ended or expired while it was in flight. + if (closed_ || !failure_.ok()) { + return; + } + if (!renewed.ok()) { + failure_ = renewed; + PAIMON_LOG_WARN(logger_, "Stopped renewing the commit lease of %s: %s", + identifier_.ToString().c_str(), renewed.ToString().c_str()); + return; + } + renewed_at_millis_ = requested_at; + } + } + + private: + Status CheckValidLocked() { + // Measured from when the last grant was requested, so the deadline also covers the + // response time, and enforced without relying on the renewal thread having run. + if (failure_.ok() && clock_() - renewed_at_millis_ >= lease_millis_) { + failure_ = Status::Invalid("the commit lease has expired locally"); + } + if (!failure_.ok()) { + return failure_.WithMessage( + fmt::format("the commit lease of {} held by commit user {} is no longer usable: {}", + identifier_.ToString(), commit_user_, failure_.message())); + } + return Status::OK(); + } + + Status Renew() const { + PAIMON_ASSIGN_OR_RAISE( + CommitLockResponse response, + api_->RenewCommitLock(identifier_, table_uuid_, commit_user_, &renewal_cancellation_)); + if (!response.IsAcquired()) { + return Status::Invalid("the commit lease has expired"); + } + PAIMON_RETURN_NOT_OK(ValidateGrant(response, commit_user_)); + if (response.GetLeaseMillis() != lease_millis_) { + return Status::Invalid(fmt::format( + "the commit lease duration changed during renewal: expected leaseMillis {}, got {}", + lease_millis_, response.GetLeaseMillis())); + } + return Status::OK(); + } + + const std::shared_ptr api_; + const Identifier identifier_; + const std::string table_uuid_; + const std::string commit_user_; + const RestCommitLock::Clock clock_; + const int64_t lease_millis_; + const std::shared_ptr logger_; + RestHttpClient::Cancellation renewal_cancellation_; + + // Guards the fields below, shared with the renewal thread. + std::mutex mutex_; + std::condition_variable closed_cv_; + bool closed_ = false; + Status failure_; + int64_t renewed_at_millis_; +}; + +// The scope of one granted lease, renewed in the background until the scope ends or the lease +// fails. +class RestCommitLease : public CommitLockScope { + public: + explicit RestCommitLease(std::shared_ptr state) : state_(std::move(state)) {} + + ~RestCommitLease() override { + state_->Close(); + } + + Status StartRenewal(const RestCommitLock::ThreadStarter& start_thread) { + // Detached rather than joined when the scope ends: a renewal can block where no + // cancellation reaches, as while refreshing credentials, and must not hold up the commit. + return start_thread([state = state_]() { state->RenewLoop(); }); + } + + Status EnsureValid() override { + return state_->EnsureValid(); + } + + private: + std::shared_ptr state_; +}; + +} // namespace + +int64_t RestCommitLock::SteadyClockMillis() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); +} + +Status RestCommitLock::StartDetachedThread(std::function task) { + try { + std::thread(std::move(task)).detach(); + } catch (const std::system_error& e) { + return Status::UnknownError(std::string("failed to start a thread: ") + e.what()); + } + return Status::OK(); +} + +RestCommitLock::RestCommitLock(const std::shared_ptr& api, const Identifier& identifier, + const std::string& table_uuid, const std::string& commit_user, + int64_t acquire_timeout_millis, Clock clock, + ThreadStarter start_thread) + : api_(api), + identifier_(identifier), + table_uuid_(table_uuid), + commit_user_(commit_user), + acquire_timeout_millis_(acquire_timeout_millis), + clock_(std::move(clock)), + start_thread_(std::move(start_thread)), + logger_(Logger::GetLogger("RestCommitLock")) {} + +Result> RestCommitLock::Create( + const std::shared_ptr& api, const Identifier& identifier, + const std::optional& table_uuid, const std::string& commit_user, + int64_t acquire_timeout_millis, Clock clock, ThreadStarter start_thread) { + if (api == nullptr || !clock || !start_thread) { + return Status::Invalid("a commit lease requires a rest api, a clock and a thread starter"); + } + if (!table_uuid || table_uuid.value().empty()) { + return Status::Invalid( + "a commit lease requires the table UUID; pass Table::CatalogUuid() to WithTableId()"); + } + if (commit_user.empty()) { + return Status::Invalid("a commit lease requires a commit user"); + } + if (acquire_timeout_millis < 0) { + return Status::Invalid("the commit lock acquisition timeout cannot be negative"); + } + return std::unique_ptr( + new RestCommitLock(api, identifier, table_uuid.value(), commit_user, acquire_timeout_millis, + std::move(clock), std::move(start_thread))); +} + +Result> RestCommitLock::Acquire() { + const int64_t started_at = clock_(); + bool waiting = false; + while (true) { + const int64_t requested_at = clock_(); + Result response = + api_->AcquireCommitLock(identifier_, table_uuid_, commit_user_); + if (!response.ok()) { + // Any error, including the rejection from a server without the lease protocol, fails + // the acquisition instead of letting the commit proceed unprotected. + const Status& status = response.status(); + return status.WithMessage(fmt::format("failed to acquire the commit lock of {}: {}", + identifier_.ToString(), status.message())); + } + const CommitLockResponse& grant = response.value(); + if (grant.IsAcquired()) { + PAIMON_RETURN_NOT_OK(ValidateGrant(grant, commit_user_)); + auto lease = std::make_unique( + std::make_shared(api_, identifier_, table_uuid_, commit_user_, clock_, + grant.GetLeaseMillis(), requested_at, logger_)); + // A grant that took longer than its lease to arrive protects nothing. + PAIMON_RETURN_NOT_OK(lease->EnsureValid()); + Status started = lease->StartRenewal(start_thread_); + if (!started.ok()) { + return started.WithMessage( + fmt::format("failed to start renewing the commit lease of {}: {}", + identifier_.ToString(), started.message())); + } + PAIMON_LOG_DEBUG(logger_, "Acquired the commit lease of %s for %ld ms", + identifier_.ToString().c_str(), grant.GetLeaseMillis()); + return std::unique_ptr(std::move(lease)); + } + const int64_t waited = clock_() - started_at; + if (waited >= acquire_timeout_millis_) { + return Status::IOError(fmt::format( + "timed out acquiring the commit lock of {} after {} ms, another writer holds it", + identifier_.ToString(), waited)); + } + if (!waiting) { + PAIMON_LOG_INFO(logger_, "Another writer holds the commit lease of %s, waiting for it", + identifier_.ToString().c_str()); + waiting = true; + } + std::this_thread::sleep_for( + std::chrono::milliseconds(std::min(kMaxPollIntervalMillis, acquire_timeout_millis_))); + } +} + +} // namespace paimon diff --git a/src/paimon/rest/rest_commit_lock.h b/src/paimon/rest/rest_commit_lock.h new file mode 100644 index 000000000..6cc94b09c --- /dev/null +++ b/src/paimon/rest/rest_commit_lock.h @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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/identifier.h" +#include "paimon/core/catalog/commit_lock.h" +#include "paimon/logging.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { +class RestApi; + +/// A renewable REST commit lease bound to a table or branch and to one writer, mirroring Java +/// `RESTCatalogLock`. +/// +/// `Acquire()` asks the server for a lease and polls while another writer holds one, until the +/// acquisition timeout elapses. The timeout is checked after each response, so it does not +/// interrupt a request in flight. The scope it returns renews the lease in the background every +/// third of the lease duration, on a thread of its own; when that thread cannot be started, +/// `Acquire()` fails and may be called again. The scope tracks the deadline on a monotonic clock +/// from when each grant was requested, so `EnsureValid()` fails once a renewal is refused or +/// fails, or the deadline passes even if no renewal ran. +/// +/// There is no unlock request. The server ends a lease when a snapshot is published under it; +/// ending a scope only stops renewing, so a lease that published nothing stays held until it +/// expires. Ending a scope does not wait for a renewal in flight: its request is cancelled, and +/// its thread exits on its own once the call returns. +class RestCommitLock : public CommitLock { + public: + /// Monotonic time in milliseconds. + using Clock = std::function; + + /// Runs a task on a new detached thread, failing instead of throwing when none can be started. + using ThreadStarter = std::function task)>; + + /// Milliseconds of `std::chrono::steady_clock`. + static int64_t SteadyClockMillis(); + + /// Runs `task` on a detached `std::thread`. + static Status StartDetachedThread(std::function task); + + /// Creates the lock without sending any request. + /// + /// @param api Client of the REST server holding the leases. + /// @param identifier Table or branch to lock, as the server addresses it. + /// @param table_uuid Catalog UUID of the table or branch, so a recreated table cannot inherit + /// a lease; required. + /// @param commit_user Exact commit user of the snapshots published under the lease; required. + /// @param acquire_timeout_millis How long `Acquire()` waits while another writer holds the + /// lease. + /// @param clock Monotonic clock, replaceable for tests. + /// @param start_thread Starts the thread renewing each lease, replaceable for tests. + static Result> Create( + const std::shared_ptr& api, const Identifier& identifier, + const std::optional& table_uuid, const std::string& commit_user, + int64_t acquire_timeout_millis, Clock clock = &RestCommitLock::SteadyClockMillis, + ThreadStarter start_thread = &RestCommitLock::StartDetachedThread); + + Result> Acquire() override; + + private: + RestCommitLock(const std::shared_ptr& api, const Identifier& identifier, + const std::string& table_uuid, const std::string& commit_user, + int64_t acquire_timeout_millis, Clock clock, ThreadStarter start_thread); + + std::shared_ptr api_; + Identifier identifier_; + std::string table_uuid_; + std::string commit_user_; + int64_t acquire_timeout_millis_; + Clock clock_; + ThreadStarter start_thread_; + std::shared_ptr logger_; +}; + +} // namespace paimon diff --git a/src/paimon/rest/rest_commit_lock_test.cpp b/src/paimon/rest/rest_commit_lock_test.cpp new file mode 100644 index 000000000..2bfa4e98d --- /dev/null +++ b/src/paimon/rest/rest_commit_lock_test.cpp @@ -0,0 +1,600 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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_commit_lock.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/catalog/identifier.h" +#include "paimon/catalog_options.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/rest/mock_rest_server.h" +#include "paimon/rest/rest_api.h" +#include "paimon/rest/rest_http_client.h" +#include "paimon/rest/rest_messages.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +namespace { + +constexpr const char kCommitUser[] = "commit-user"; +constexpr const char kTableUuid[] = "table-uuid"; +constexpr const char kAcquirePath[] = "/v1/databases/db/tables/t1/commit-lock"; +constexpr const char kRenewPath[] = "/v1/databases/db/tables/t1/commit-lock/renew"; + +struct LeaseServerState { + std::mutex mutex; + std::vector acquire_requests; + std::vector renew_requests; + int32_t contended_acquires = 0; + int64_t lease_millis = 60000; + std::optional granted_owner; + std::optional renewed_lease_millis; + size_t refused_renewals = 0; + std::set refused_renewal_users; + std::optional acquire_error_code; + std::optional renew_error_code; + int64_t acquire_delay_millis = 0; + std::optional answered_renewals; + std::condition_variable renewal_answered; + std::atomic now_millis{1000}; +}; + +MockRestServer::Response JsonResponse(int32_t code, const std::string& body) { + MockRestServer::Response response; + response.code = code; + response.body = body; + return response; +} + +MockRestServer::Response Error(int32_t code) { + ErrorResponse error(ErrorResponse::kResourceTypeTable, "t1", "lease request rejected", code); + return JsonResponse(code, error.ToJsonString().value()); +} + +MockRestServer::Response Lease(bool acquired, const std::string& owner, int64_t lease_millis) { + CommitLockResponse response( + acquired, acquired ? std::optional(owner) : std::nullopt, + acquired ? 1700000000000 + lease_millis : 0, acquired ? lease_millis : 0, std::nullopt); + return JsonResponse(200, response.ToJsonString().value()); +} + +MockRestServer::Response HandleLeaseRequest(LeaseServerState* state, + const MockRestServer::Request& request) { + std::unique_lock lock(state->mutex); + if (request.method != "POST") { + return Error(400); + } + Result parsed = CommitLockRequest::FromJsonString(request.body); + if (!parsed.ok()) { + return Error(400); + } + if (request.path == kAcquirePath) { + state->acquire_requests.push_back(parsed.value()); + state->now_millis += state->acquire_delay_millis; + if (state->acquire_error_code) { + return Error(state->acquire_error_code.value()); + } + if (state->contended_acquires > 0) { + --state->contended_acquires; + return Lease(false, "", 0); + } + return Lease(true, state->granted_owner.value_or(parsed.value().GetCommitUser()), + state->lease_millis); + } + if (request.path == kRenewPath) { + state->renew_requests.push_back(parsed.value()); + const size_t renewal = state->renew_requests.size(); + state->renewal_answered.wait_for(lock, std::chrono::seconds(10), [state, renewal]() { + return !state->answered_renewals || state->answered_renewals.value() >= renewal; + }); + if (state->renew_error_code) { + return Error(state->renew_error_code.value()); + } + if (renewal <= state->refused_renewals || + state->refused_renewal_users.count(parsed.value().GetCommitUser()) > 0) { + return Lease(false, "", 0); + } + return Lease(true, state->granted_owner.value_or(parsed.value().GetCommitUser()), + state->renewed_lease_millis.value_or(state->lease_millis)); + } + return Error(404); +} + +} // namespace + +class RestCommitLockTest : 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 HandleLeaseRequest(state.get(), req); + })); + std::map options = { + {CatalogOptions::URI, server_->GetBaseUri()}, + {CatalogOptions::TOKEN_PROVIDER, "bear"}, + {CatalogOptions::TOKEN, "test-token"}, + }; + RestHttpClient::Config config; + config.retry_base_delay_ms = 1; + ASSERT_OK_AND_ASSIGN(std::unique_ptr api, + RestApi::Create(options, "", /*config_required=*/false, config)); + api_ = std::move(api); + } + + void TearDown() override { + if (server_) { + server_->Stop(); + } + } + + Result> CreateLock( + int64_t acquire_timeout_millis = 10000, const std::string& commit_user = kCommitUser, + RestCommitLock::ThreadStarter start_thread = &RestCommitLock::StartDetachedThread) { + std::shared_ptr state = state_; + return RestCommitLock::Create( + api_, Identifier("db", "t1"), std::string(kTableUuid), commit_user, + acquire_timeout_millis, [state]() { return state->now_millis.load(); }, + std::move(start_thread)); + } + + size_t AcquireRequests() { + std::lock_guard lock(state_->mutex); + return state_->acquire_requests.size(); + } + + size_t RenewRequests(const std::optional& commit_user = std::nullopt) { + std::lock_guard lock(state_->mutex); + return static_cast( + std::count_if(state_->renew_requests.begin(), state_->renew_requests.end(), + [&commit_user](const CommitLockRequest& request) { + return !commit_user || request.GetCommitUser() == commit_user.value(); + })); + } + + bool AwaitRenewals(size_t count, const std::optional& commit_user = std::nullopt) { + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (RenewRequests(commit_user) < count) { + if (std::chrono::steady_clock::now() > deadline) { + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + return true; + } + + void AnswerRenewals(std::optional count) { + { + std::lock_guard lock(state_->mutex); + state_->answered_renewals = count; + } + state_->renewal_answered.notify_all(); + } + + static Status AwaitInvalid(CommitLockScope* scope) { + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (true) { + Status status = scope->EnsureValid(); + if (!status.ok() || std::chrono::steady_clock::now() > deadline) { + return status; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + } + + std::shared_ptr state_; + std::unique_ptr server_; + std::shared_ptr api_; +}; + +TEST_F(RestCommitLockTest, MissingIdentityIsRejectedBeforeAnyRequest) { + Identifier identifier("db", "t1"); + ASSERT_NOK_WITH_MSG(RestCommitLock::Create(api_, identifier, std::nullopt, kCommitUser, + /*acquire_timeout_millis=*/0), + "requires the table UUID"); + ASSERT_NOK_WITH_MSG(RestCommitLock::Create(api_, identifier, std::string(), kCommitUser, 0), + "requires the table UUID"); + ASSERT_NOK_WITH_MSG(RestCommitLock::Create(api_, identifier, std::string(kTableUuid), "", 0), + "requires a commit user"); + ASSERT_NOK_WITH_MSG( + RestCommitLock::Create(api_, identifier, std::string(kTableUuid), kCommitUser, -1), + "cannot be negative"); + ASSERT_NOK_WITH_MSG( + RestCommitLock::Create(nullptr, identifier, std::string(kTableUuid), kCommitUser, 0), + "requires a rest api"); + ASSERT_NOK_WITH_MSG( + RestCommitLock::Create(api_, identifier, std::string(kTableUuid), kCommitUser, 0, + &RestCommitLock::SteadyClockMillis, /*start_thread=*/nullptr), + "a thread starter"); + ASSERT_EQ(0u, AcquireRequests()); +} + +TEST_F(RestCommitLockTest, AcquirePollsUntilGranted) { + { + std::lock_guard guard(state_->mutex); + state_->contended_acquires = 2; + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock, CreateLock()); + ASSERT_EQ(0u, AcquireRequests()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, lock->Acquire()); + ASSERT_NE(nullptr, scope); + ASSERT_OK(scope->EnsureValid()); + std::lock_guard guard(state_->mutex); + ASSERT_EQ(3u, state_->acquire_requests.size()); + for (const CommitLockRequest& request : state_->acquire_requests) { + ASSERT_EQ(kTableUuid, request.GetTableId()); + ASSERT_EQ(kCommitUser, request.GetCommitUser()); + } +} + +TEST_F(RestCommitLockTest, ContentionTimesOutWithoutRenewing) { + { + std::lock_guard guard(state_->mutex); + state_->contended_acquires = 1000; + state_->acquire_delay_millis = 50; + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock, + CreateLock(/*acquire_timeout_millis=*/120)); + Status status = lock->Acquire().status(); + ASSERT_TRUE(status.IsIOError()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "timed out acquiring the commit lock"); + ASSERT_EQ(3u, AcquireRequests()); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr impatient, + CreateLock(/*acquire_timeout_millis=*/0)); + ASSERT_NOK_WITH_MSG(impatient->Acquire(), "timed out acquiring the commit lock"); + ASSERT_EQ(4u, AcquireRequests()); + ASSERT_EQ(0u, RenewRequests()); +} + +TEST_F(RestCommitLockTest, FailedAcquisitionStartsNoLease) { + auto failed_starts = std::make_shared>(1); + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock, + CreateLock(/*acquire_timeout_millis=*/10000, kCommitUser, + [failed_starts](std::function task) -> Status { + if (failed_starts->fetch_sub(1) > 0) { + return Status::UnknownError("no thread available"); + } + return RestCommitLock::StartDetachedThread(std::move(task)); + })); + for (int32_t code : {400, 403, 404}) { + SCOPED_TRACE(code); + { + std::lock_guard guard(state_->mutex); + state_->acquire_error_code = code; + } + Status status = lock->Acquire().status(); + ASSERT_NOK_WITH_MSG(status, "failed to acquire the commit lock of"); + ASSERT_NOK_WITH_MSG(status, "lease request rejected"); + if (code == 400) { + ASSERT_TRUE(status.IsInvalid()) << status.ToString(); + } else if (code == 403) { + ASSERT_TRUE(status.IsIOError()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "forbidden"); + } else { + ASSERT_TRUE(status.IsNotExist()) << status.ToString(); + } + } + + for (bool other_owner : {true, false}) { + SCOPED_TRACE(other_owner); + { + std::lock_guard guard(state_->mutex); + state_->acquire_error_code.reset(); + state_->granted_owner = other_owner ? "another-writer" : kCommitUser; + state_->lease_millis = other_owner ? 60000 : 0; + } + Status status = lock->Acquire().status(); + ASSERT_TRUE(status.IsInvalid()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "the server returned an invalid commit lease"); + const char* reason = other_owner ? "expected owner 'commit-user', got 'another-writer'" + : "expected a positive leaseMillis, got 0"; + ASSERT_NOK_WITH_MSG(status, reason); + } + + { + std::lock_guard guard(state_->mutex); + state_->granted_owner.reset(); + state_->lease_millis = 30; + } + Status status = lock->Acquire().status(); + ASSERT_TRUE(status.IsUnknownError()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "failed to start renewing the commit lease of"); + ASSERT_NOK_WITH_MSG(status, "no thread available"); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + ASSERT_EQ(6u, AcquireRequests()); + ASSERT_EQ(0u, RenewRequests()); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, lock->Acquire()); + ASSERT_TRUE(AwaitRenewals(2)); + ASSERT_OK(scope->EnsureValid()); + ASSERT_EQ(7u, AcquireRequests()); +} + +TEST_F(RestCommitLockTest, RenewalExtendsTheLeaseFromEachRequest) { + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 30; + state_->answered_renewals = 0; + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock, CreateLock()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, lock->Acquire()); + ASSERT_TRUE(AwaitRenewals(1)); + state_->now_millis += 20; + AnswerRenewals(1); + ASSERT_TRUE(AwaitRenewals(2)); + state_->now_millis += 9; + AnswerRenewals(2); + ASSERT_TRUE(AwaitRenewals(3)); + state_->now_millis += 20; + ASSERT_OK(scope->EnsureValid()); + state_->now_millis += 1; + ASSERT_NOK_WITH_MSG(scope->EnsureValid(), "expired locally"); + AnswerRenewals(3); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + ASSERT_NOK_WITH_MSG(scope->EnsureValid(), "expired locally"); + ASSERT_EQ(3u, RenewRequests()); + std::lock_guard guard(state_->mutex); + for (const CommitLockRequest& renewal : state_->renew_requests) { + ASSERT_EQ(kTableUuid, renewal.GetTableId()); + ASSERT_EQ(kCommitUser, renewal.GetCommitUser()); + } +} + +TEST_F(RestCommitLockTest, EndingTheScopeDoesNotWaitForARenewalInFlight) { + struct TokenServerState { + std::mutex mutex; + std::condition_variable changed; + bool hold = false; + bool released = false; + int32_t held = 0; + }; + auto tokens = std::make_shared(); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr token_server, + MockRestServer::Start([tokens](const MockRestServer::Request&) { + std::unique_lock lock(tokens->mutex); + if (tokens->hold) { + ++tokens->held; + tokens->changed.notify_all(); + tokens->changed.wait_for(lock, std::chrono::seconds(30), + [&tokens]() { return tokens->released; }); + } + MockRestServer::Response response; + response.body = R"({"AccessKeyId": "ak", "AccessKeySecret": "sk",)" + R"( "SecurityToken": "sts", "Expiration": "2020-01-01T00:00:00Z"})"; + return response; + })); + auto release_tokens = [tokens]() { + { + std::lock_guard lock(tokens->mutex); + tokens->released = true; + } + tokens->changed.notify_all(); + }; + ScopeGuard release_on_exit(release_tokens); + + std::map options = { + {CatalogOptions::URI, server_->GetBaseUri()}, + {CatalogOptions::TOKEN_PROVIDER, "dlf"}, + {CatalogOptions::DLF_REGION, "cn-hangzhou"}, + {CatalogOptions::DLF_TOKEN_LOADER, "ecs"}, + {CatalogOptions::DLF_TOKEN_ECS_METADATA_URL, token_server->GetBaseUri() + "/"}, + {CatalogOptions::DLF_TOKEN_ECS_ROLE_NAME, "role"}, + }; + ASSERT_OK_AND_ASSIGN(std::unique_ptr dlf_api, + RestApi::Create(options, "", /*config_required=*/false)); + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 30; + } + std::shared_ptr state = state_; + ASSERT_OK_AND_ASSIGN( + std::unique_ptr lock, + RestCommitLock::Create(std::move(dlf_api), Identifier("db", "t1"), std::string(kTableUuid), + kCommitUser, /*acquire_timeout_millis=*/10000, + [state]() { return state->now_millis.load(); })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, lock->Acquire()); + { + std::unique_lock guard(tokens->mutex); + tokens->hold = true; + ASSERT_TRUE(tokens->changed.wait_for(guard, std::chrono::seconds(10), + [&tokens]() { return tokens->held > 0; })); + } + size_t renewals = RenewRequests(); + auto ended_at = std::chrono::steady_clock::now(); + scope.reset(); + std::chrono::steady_clock::duration ending = std::chrono::steady_clock::now() - ended_at; + release_tokens(); + ASSERT_LT(ending, std::chrono::seconds(1)); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + ASSERT_EQ(renewals, RenewRequests()); +} + +TEST_F(RestCommitLockTest, LateAnswerToAnEndedScopeCannotAffectTheNextScope) { + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 30; + state_->answered_renewals = 0; + state_->refused_renewals = 1; + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock, CreateLock()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr first, lock->Acquire()); + ASSERT_TRUE(AwaitRenewals(1)); + first.reset(); + + Result> second = Status::UnknownError("not acquired yet"); + std::thread acquiring([&lock, &second]() { second = lock->Acquire(); }); + ScopeGuard join_on_exit([&acquiring]() { + if (acquiring.joinable()) { + acquiring.join(); + } + }); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + AnswerRenewals(std::nullopt); + acquiring.join(); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, std::move(second)); + ASSERT_TRUE(AwaitRenewals(3)); + ASSERT_OK(scope->EnsureValid()); + ASSERT_EQ(2u, AcquireRequests()); +} + +TEST_F(RestCommitLockTest, LeasesOfDifferentWritersDoNotInterfere) { + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 30; + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock_a, + CreateLock(/*acquire_timeout_millis=*/10000, "writer-a")); + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock_b, + CreateLock(/*acquire_timeout_millis=*/10000, "writer-b")); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope_a, lock_a->Acquire()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope_b, lock_b->Acquire()); + + ASSERT_TRUE(AwaitRenewals(1, "writer-b")); + scope_b.reset(); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + size_t renewals_b = RenewRequests("writer-b"); + ASSERT_TRUE(AwaitRenewals(RenewRequests("writer-a") + 2, "writer-a")); + ASSERT_EQ(renewals_b, RenewRequests("writer-b")); + ASSERT_OK(scope_a->EnsureValid()); + + ASSERT_OK_AND_ASSIGN(scope_b, lock_b->Acquire()); + { + std::lock_guard guard(state_->mutex); + state_->refused_renewal_users.insert("writer-b"); + } + ASSERT_NOK_WITH_MSG(AwaitInvalid(scope_b.get()), + "held by commit user writer-b is no longer usable"); + ASSERT_TRUE(AwaitRenewals(RenewRequests("writer-a") + 2, "writer-a")); + ASSERT_OK(scope_a->EnsureValid()); +} + +TEST_F(RestCommitLockTest, FailedRenewalInvalidatesTheLease) { + enum class Failure { kLost, kError, kOtherOwner, kOtherDuration }; + for (Failure failure : + {Failure::kLost, Failure::kError, Failure::kOtherOwner, Failure::kOtherDuration}) { + SCOPED_TRACE(static_cast(failure)); + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 30; + state_->refused_renewal_users.clear(); + state_->renew_error_code.reset(); + state_->granted_owner = kCommitUser; + state_->renewed_lease_millis.reset(); + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock, CreateLock()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, lock->Acquire()); + { + std::lock_guard guard(state_->mutex); + if (failure == Failure::kLost) { + state_->refused_renewal_users.insert(kCommitUser); + } + state_->renew_error_code = + failure == Failure::kError ? std::optional(500) : std::nullopt; + state_->granted_owner = + failure == Failure::kOtherOwner ? "another-writer" : kCommitUser; + state_->renewed_lease_millis = + failure == Failure::kOtherDuration ? std::optional(60) : std::nullopt; + } + Status status = AwaitInvalid(scope.get()); + ASSERT_NOK_WITH_MSG(status, "no longer usable"); + switch (failure) { + case Failure::kLost: + ASSERT_TRUE(status.IsInvalid()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "the commit lease has expired"); + break; + case Failure::kError: + ASSERT_TRUE(status.IsIOError()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "server error"); + break; + case Failure::kOtherOwner: + ASSERT_TRUE(status.IsInvalid()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "expected owner 'commit-user', got 'another-writer'"); + break; + case Failure::kOtherDuration: + ASSERT_TRUE(status.IsInvalid()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, + "duration changed during renewal: expected leaseMillis 30, " + "got 60"); + break; + } + size_t renewals = RenewRequests(); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + ASSERT_EQ(renewals, RenewRequests()); + ASSERT_NOK(scope->EnsureValid()); + } +} + +TEST_F(RestCommitLockTest, LocalExpiryInvalidatesTheLease) { + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 1000; + state_->acquire_delay_millis = 1000; + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr lock, CreateLock()); + ASSERT_NOK_WITH_MSG(lock->Acquire(), "expired locally"); + + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 60000; + state_->acquire_delay_millis = 0; + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr scope, lock->Acquire()); + state_->now_millis += 59999; + ASSERT_OK(scope->EnsureValid()); + state_->now_millis += 1; + Status status = scope->EnsureValid(); + ASSERT_TRUE(status.IsInvalid()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "expired locally"); + state_->now_millis -= 30000; + ASSERT_NOK_WITH_MSG(scope->EnsureValid(), "expired locally"); + ASSERT_EQ(0u, RenewRequests()); + + { + std::lock_guard guard(state_->mutex); + state_->lease_millis = 30; + state_->answered_renewals = 0; + } + ASSERT_OK_AND_ASSIGN(scope, lock->Acquire()); + ASSERT_TRUE(AwaitRenewals(1)); + state_->now_millis += 30; + AnswerRenewals(1); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + ASSERT_EQ(1u, RenewRequests()); + ASSERT_NOK_WITH_MSG(scope->EnsureValid(), "expired locally"); +} + +} // namespace paimon::test diff --git a/src/paimon/rest/rest_http_client.cpp b/src/paimon/rest/rest_http_client.cpp index e78c62fee..6c3fe7e9a 100644 --- a/src/paimon/rest/rest_http_client.cpp +++ b/src/paimon/rest/rest_http_client.cpp @@ -134,6 +134,17 @@ bool IsRetriableTransportError(CURLcode code) { } } +// A non-zero result makes libcurl abort the transfer with CURLE_ABORTED_BY_CALLBACK. +int AbortWhenCancelled(void* user_data, curl_off_t /*dltotal*/, curl_off_t /*dlnow*/, + curl_off_t /*ultotal*/, curl_off_t /*ulnow*/) { + const auto* cancellation = static_cast(user_data); + return cancellation->IsCancelled() ? 1 : 0; +} + +Status CancelledStatus(const std::string& method) { + return Status::Cancelled(fmt::format("http {} request was cancelled", method)); +} + // libcurl sends no User-Agent of its own, so requests would reach the server unnamed. A // "header.User-Agent" option still wins: an explicit request header overrides // CURLOPT_USERAGENT. @@ -202,6 +213,24 @@ class RestHttpClient::HandlePool { mutable std::vector handles_; }; +void RestHttpClient::Cancellation::Cancel() { + { + std::lock_guard lock(mutex_); + cancelled_.store(true); + } + cancelled_cv_.notify_all(); +} + +bool RestHttpClient::Cancellation::IsCancelled() const { + return cancelled_.load(); +} + +void RestHttpClient::Cancellation::WaitFor(int64_t delay_ms) const { + std::unique_lock lock(mutex_); + cancelled_cv_.wait_for(lock, std::chrono::milliseconds(delay_ms), + [this]() { return cancelled_.load(); }); +} + RestHttpClient::RestHttpClient(const std::string& base_uri, const Config& config) : handle_pool_(std::make_unique()), base_uri_(base_uri), @@ -252,7 +281,7 @@ std::string RestHttpClient::BuildQueryString( Result RestHttpClient::ExecuteOnce( const std::string& method, const std::string& url, const std::map& headers, const std::string& body, - bool follow_redirects, bool* transport_retriable) const { + bool follow_redirects, const Cancellation* cancellation, bool* transport_retriable) const { CURL* curl = handle_pool_->Acquire(); if (curl == nullptr) { return Status::IOError("failed to create curl handle"); @@ -274,6 +303,13 @@ Result RestHttpClient::ExecuteOnce( curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response.body); curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, WriteHeaderCallback); curl_easy_setopt(curl, CURLOPT_HEADERDATA, &response.headers); + if (cancellation != nullptr) { + // The callback only reads the cancellation; curl just takes a non-const pointer. + curl_easy_setopt(curl, CURLOPT_XFERINFOFUNCTION, AbortWhenCancelled); + curl_easy_setopt(curl, CURLOPT_XFERINFODATA, + const_cast(static_cast(cancellation))); + curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L); + } // Redirects can be disabled for auth headers whose signatures are bound to the // original request. Otherwise they are restricted to http(s) targets. Without // CURLOPT_POSTREDIR a 301/302 would replay a body-carrying request as a bodyless @@ -325,6 +361,9 @@ Result RestHttpClient::ExecuteOnce( curl_easy_setopt(curl, CURLOPT_HTTPHEADER, header_list); CURLcode curl_code = curl_easy_perform(curl); + if (curl_code == CURLE_ABORTED_BY_CALLBACK && cancellation != nullptr) { + return CancelledStatus(method); + } if (curl_code != CURLE_OK) { *transport_retriable = IsRetriableTransportError(curl_code); // Neither the url nor the response body is echoed in errors: both may carry @@ -404,7 +443,7 @@ Result RestHttpClient::Execute( const std::string& method, const std::string& path, const std::map& query_params, const std::map& headers, const std::string& body, - bool follow_redirects, bool retry_safe) const { + bool follow_redirects, bool retry_safe, const Cancellation* cancellation) const { if (method != "GET" && method != "POST" && method != "DELETE") { return Status::Invalid(fmt::format("unsupported http method: {}", method)); } @@ -428,10 +467,14 @@ Result RestHttpClient::Execute( std::chrono::steady_clock::time_point start = std::chrono::steady_clock::now(); int32_t execution_count = 0; while (true) { + // Also ends a retry backoff that the cancellation cut short. + if (cancellation != nullptr && cancellation->IsCancelled()) { + return CancelledStatus(method); + } execution_count++; bool transport_retriable = false; - Result result = - ExecuteOnce(method, url, headers, body, follow_redirects, &transport_retriable); + Result result = ExecuteOnce(method, url, headers, body, follow_redirects, + cancellation, &transport_retriable); bool retriable; if (result.ok()) { retriable = IsRetriableCode(result.value().code); @@ -465,7 +508,11 @@ Result RestHttpClient::Execute( static_cast(delay_ms.value()), // NOLINT(runtime/int) execution_count, config_.max_retries, reason.c_str()); if (delay_ms.value() > 0) { - std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms.value())); + if (cancellation != nullptr) { + cancellation->WaitFor(delay_ms.value()); + } else { + std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms.value())); + } } } } diff --git a/src/paimon/rest/rest_http_client.h b/src/paimon/rest/rest_http_client.h index 60bbb0d0f..b01c6e76d 100644 --- a/src/paimon/rest/rest_http_client.h +++ b/src/paimon/rest/rest_http_client.h @@ -18,9 +18,12 @@ #pragma once +#include +#include #include #include #include +#include #include #include @@ -82,6 +85,25 @@ class RestHttpClient { } }; + /// Lets one thread abandon the `Execute` calls another thread runs with it. A retry + /// backoff ends at once. A transfer stops within about a second: libcurl checks for the + /// cancellation in its progress callback, which it calls about once a second while no + /// data moves. A request already sent may still reach the server. + class Cancellation { + public: + void Cancel(); + + bool IsCancelled() const; + + /// Waits up to `delay_ms`, returning early once cancelled. + void WaitFor(int64_t delay_ms) const; + + private: + mutable std::mutex mutex_; + mutable std::condition_variable cancelled_cv_; + std::atomic cancelled_{false}; + }; + /// Creates a client against `base_uri`. The uri is normalized: surrounding /// whitespace and all trailing '/' are stripped and "http://" is prepended when no /// scheme is present. @@ -103,11 +125,16 @@ class RestHttpClient { /// /// `retry_safe=false` disables retries and redirects, overriding `follow_redirects`, /// to prevent replay when the original request's outcome is unknown. + /// + /// Once `cancellation` is cancelled, the call returns a `Cancelled` status instead of + /// waiting for a response, starting another attempt or finishing a retry backoff; null + /// leaves the call bounded only by its timeouts. Result Execute(const std::string& method, const std::string& path, const std::map& query_params, const std::map& headers, const std::string& body, bool follow_redirects = true, - bool retry_safe = true) const; + bool retry_safe = true, + const Cancellation* cancellation = nullptr) const; const std::string& GetBaseUri() const { return base_uri_; @@ -143,7 +170,7 @@ class RestHttpClient { Result ExecuteOnce(const std::string& method, const std::string& url, const std::map& headers, const std::string& body, bool follow_redirects, - bool* transport_retriable) const; + const Cancellation* cancellation, bool* transport_retriable) const; std::optional GetRetryDelayMs(int32_t execution_count, const Response* response, int64_t remaining_budget_ms) const; diff --git a/src/paimon/rest/rest_http_client_test.cpp b/src/paimon/rest/rest_http_client_test.cpp index d03005f5c..28d7c63dc 100644 --- a/src/paimon/rest/rest_http_client_test.cpp +++ b/src/paimon/rest/rest_http_client_test.cpp @@ -20,12 +20,15 @@ #include #include +#include #include #include #include +#include #include #include "gtest/gtest.h" +#include "paimon/common/utils/scope_guard.h" #include "paimon/rest/mock_rest_server.h" #include "paimon/testing/utils/testharness.h" @@ -623,6 +626,125 @@ TEST(RestHttpClientTest, InvalidHeadersAreRejected) { "invalid http header value for 'X-Ok'"); } +namespace { + +struct ResponseGate { + void AwaitRequests(int32_t count) { + std::unique_lock lock(mutex); + changed.wait_for(lock, std::chrono::seconds(10), [&]() { return requests >= count; }); + } + + void Release() { + { + std::lock_guard lock(mutex); + released = true; + } + changed.notify_all(); + } + + MockRestServer::Response Hold() { + std::unique_lock lock(mutex); + ++requests; + changed.notify_all(); + changed.wait_for(lock, std::chrono::seconds(30), [&]() { return released; }); + MockRestServer::Response response; + response.close_without_response = true; + return response; + } + + std::mutex mutex; + std::condition_variable changed; + int32_t requests = 0; + bool released = false; +}; + +} // namespace + +TEST(RestHttpClientTest, CancellationStopsWaitingForAResponse) { + ResponseGate gate; + ASSERT_OK_AND_ASSIGN( + std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request&) { return gate.Hold(); })); + ScopeGuard release([&]() { gate.Release(); }); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + RestHttpClient::Create(server->GetBaseUri(), FastRetryConfig(5))); + RestHttpClient::Cancellation cancellation; + Status status; + std::thread request([&]() { + status = client + ->Execute("POST", "/v1/slow", {}, {}, "{}", /*follow_redirects=*/true, + /*retry_safe=*/true, &cancellation) + .status(); + }); + gate.AwaitRequests(1); + auto cancelled_at = std::chrono::steady_clock::now(); + cancellation.Cancel(); + request.join(); + ASSERT_LT(std::chrono::steady_clock::now() - cancelled_at, std::chrono::seconds(10)); + ASSERT_TRUE(status.IsCancelled()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "http POST request was cancelled"); + std::lock_guard lock(gate.mutex); + ASSERT_EQ(1, gate.requests); +} + +TEST(RestHttpClientTest, CancellationEndsARetryBackoff) { + std::atomic request_count{0}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request&) { + ++request_count; + MockRestServer::Response response; + response.code = 503; + response.headers["Retry-After"] = "60"; + return response; + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + RestHttpClient::Create(server->GetBaseUri(), FastRetryConfig(5))); + RestHttpClient::Cancellation cancellation; + Status status; + std::thread request([&]() { + status = client + ->Execute("POST", "/v1/busy", {}, {}, "{}", /*follow_redirects=*/true, + /*retry_safe=*/true, &cancellation) + .status(); + }); + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (request_count.load() == 0 && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + auto cancelled_at = std::chrono::steady_clock::now(); + cancellation.Cancel(); + request.join(); + ASSERT_LT(std::chrono::steady_clock::now() - cancelled_at, std::chrono::seconds(10)); + ASSERT_TRUE(status.IsCancelled()) << status.ToString(); + ASSERT_EQ(1, request_count.load()); +} + +TEST(RestHttpClientTest, CancelledRequestIsNotSent) { + std::atomic request_count{0}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr server, + MockRestServer::Start([&](const MockRestServer::Request&) { + ++request_count; + return MockRestServer::Response(); + })); + ASSERT_OK_AND_ASSIGN(std::unique_ptr client, + RestHttpClient::Create(server->GetBaseUri())); + RestHttpClient::Cancellation cancellation; + cancellation.Cancel(); + Status status = client + ->Execute("GET", "/v1/config", {}, {}, "", /*follow_redirects=*/true, + /*retry_safe=*/true, &cancellation) + .status(); + ASSERT_TRUE(status.IsCancelled()) << status.ToString(); + + RestHttpClient::Cancellation not_cancelled; + ASSERT_OK_AND_ASSIGN(RestHttpClient::Response response, + client->Execute("GET", "/v1/config", {}, {}, "", /*follow_redirects=*/true, + /*retry_safe=*/true, ¬_cancelled)); + ASSERT_EQ(200, response.code); + ASSERT_EQ(1, request_count.load()); +} + TEST(RestHttpClientTest, UnsupportedMethod) { ASSERT_OK_AND_ASSIGN(std::unique_ptr client, RestHttpClient::Create("http://127.0.0.1:1")); diff --git a/src/paimon/rest/rest_messages.cpp b/src/paimon/rest/rest_messages.cpp index f9b7048c2..177b26d9e 100644 --- a/src/paimon/rest/rest_messages.cpp +++ b/src/paimon/rest/rest_messages.cpp @@ -58,6 +58,10 @@ constexpr const char kFieldSuccess[] = "success"; constexpr const char kFieldSnapshot[] = "snapshot"; constexpr const char kFieldToken[] = "token"; constexpr const char kFieldExpiresAtMillis[] = "expiresAtMillis"; +constexpr const char kFieldTableId[] = "tableId"; +constexpr const char kFieldCommitUser[] = "commitUser"; +constexpr const char kFieldAcquired[] = "acquired"; +constexpr const char kFieldLeaseMillis[] = "leaseMillis"; void AddOptionalStringMember(rapidjson::Value* obj, const char* key, const std::optional& value, @@ -453,4 +457,45 @@ void CommitTableResponse::FromJson(const rapidjson::Value& obj) noexcept(false) success_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldSuccess); } +rapidjson::Value CommitLockRequest::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldTableId), + RapidJsonUtil::SerializeValue(table_id_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldCommitUser), + RapidJsonUtil::SerializeValue(commit_user_, allocator).Move(), *allocator); + return obj; +} + +void CommitLockRequest::FromJson(const rapidjson::Value& obj) noexcept(false) { + table_id_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldTableId); + commit_user_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldCommitUser); +} + +rapidjson::Value CommitLockResponse::ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) { + rapidjson::Value obj(rapidjson::kObjectType); + obj.AddMember(rapidjson::StringRef(kFieldAcquired), + RapidJsonUtil::SerializeValue(acquired_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldCommitUser), + RapidJsonUtil::SerializeValue(commit_user_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldExpiresAtMillis), + RapidJsonUtil::SerializeValue(expires_at_millis_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldLeaseMillis), + RapidJsonUtil::SerializeValue(lease_millis_, allocator).Move(), *allocator); + obj.AddMember(rapidjson::StringRef(kFieldSnapshot), + RapidJsonUtil::SerializeValue(snapshot_, allocator).Move(), *allocator); + return obj; +} + +void CommitLockResponse::FromJson(const rapidjson::Value& obj) noexcept(false) { + // Required: a response that says nothing about the lease must not read as contention. + acquired_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldAcquired); + commit_user_ = + RapidJsonUtil::DeserializeKeyValue>(obj, kFieldCommitUser); + expires_at_millis_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldExpiresAtMillis, 0); + lease_millis_ = RapidJsonUtil::DeserializeKeyValue(obj, kFieldLeaseMillis, 0); + snapshot_ = RapidJsonUtil::DeserializeKeyValue>(obj, kFieldSnapshot); +} + } // namespace paimon diff --git a/src/paimon/rest/rest_messages.h b/src/paimon/rest/rest_messages.h index 76c47b8fc..1f01cb8db 100644 --- a/src/paimon/rest/rest_messages.h +++ b/src/paimon/rest/rest_messages.h @@ -458,4 +458,78 @@ class CommitTableResponse : public Jsonizable { bool success_ = false; }; +/// Asks for, or renews, the commit lease of a table or branch for one writer. +class CommitLockRequest : public Jsonizable { + public: + /// @param table_id Catalog UUID of the table or branch; a recreated table cannot inherit a + /// lease. + /// @param commit_user Exact commit user of the snapshots published under the lease. + CommitLockRequest(const std::string& table_id, const std::string& commit_user) + : table_id_(table_id), commit_user_(commit_user) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + const std::string& GetTableId() const { + return table_id_; + } + const std::string& GetCommitUser() const { + return commit_user_; + } + + CommitLockRequest() = default; + + private: + std::string table_id_; + std::string commit_user_; +}; + +/// Outcome of acquiring or renewing a commit lease. Contention, a lost lease and an expired lease +/// are all reported as `acquired=false`. +class CommitLockResponse : public Jsonizable { + public: + CommitLockResponse(bool acquired, const std::optional& commit_user, + int64_t expires_at_millis, int64_t lease_millis, + const std::optional& snapshot) + : acquired_(acquired), + commit_user_(commit_user), + expires_at_millis_(expires_at_millis), + lease_millis_(lease_millis), + snapshot_(snapshot) {} + + rapidjson::Value ToJson(rapidjson::Document::AllocatorType* allocator) const + noexcept(false) override; + void FromJson(const rapidjson::Value& obj) noexcept(false) override; + + bool IsAcquired() const { + return acquired_; + } + /// Owner of the lease, or null when it was not acquired. + const std::optional& GetCommitUser() const { + return commit_user_; + } + /// Expiration on the server's clock; zero when the lease was not acquired. + int64_t GetExpiresAtMillis() const { + return expires_at_millis_; + } + /// Lease duration, fixed within a scope; zero when the lease was not acquired. + int64_t GetLeaseMillis() const { + return lease_millis_; + } + /// Head at grant time; null for an empty table. + const std::optional& GetSnapshot() const { + return snapshot_; + } + + CommitLockResponse() = default; + + private: + bool acquired_ = false; + std::optional commit_user_; + int64_t expires_at_millis_ = 0; + int64_t lease_millis_ = 0; + std::optional snapshot_; +}; + } // namespace paimon diff --git a/src/paimon/rest/rest_messages_test.cpp b/src/paimon/rest/rest_messages_test.cpp index 3a9726d31..627b394d0 100644 --- a/src/paimon/rest/rest_messages_test.cpp +++ b/src/paimon/rest/rest_messages_test.cpp @@ -19,6 +19,7 @@ #include "paimon/rest/rest_messages.h" #include +#include #include #include "gtest/gtest.h" @@ -375,6 +376,74 @@ TEST(RestMessagesTest, CommitTableResponseRoundTrip) { ASSERT_NOK(CommitTableResponse::FromJsonString("not json")); } +TEST(RestMessagesTest, CommitLockRequestRoundTrip) { + CommitLockRequest request("table-uuid", "commit-user"); + ASSERT_OK_AND_ASSIGN(std::string json, request.ToJsonString()); + rapidjson::Document doc; + doc.Parse(json.c_str()); + ASSERT_FALSE(doc.HasParseError()); + ASSERT_EQ(2u, doc.MemberCount()); + ASSERT_STREQ("table-uuid", doc["tableId"].GetString()); + ASSERT_STREQ("commit-user", doc["commitUser"].GetString()); + + ASSERT_OK_AND_ASSIGN(CommitLockRequest parsed, CommitLockRequest::FromJsonString(json)); + ASSERT_EQ("table-uuid", parsed.GetTableId()); + ASSERT_EQ("commit-user", parsed.GetCommitUser()); + + ASSERT_NOK_WITH_MSG(CommitLockRequest::FromJsonString(R"({"commitUser": "u"})"), + "key must exist"); + ASSERT_NOK_WITH_MSG(CommitLockRequest::FromJsonString(R"({"tableId": "t"})"), "key must exist"); +} + +TEST(RestMessagesTest, CommitLockResponseRoundTrip) { + std::string granted_json = + R"({"acquired": true, "commitUser": "commit-user", "expiresAtMillis": 1700000060000,)" + R"( "leaseMillis": 60000, "snapshot": {"version": 3, "id": 7, "schemaId": 0,)" + R"( "baseManifestList": "bml", "deltaManifestList": "dml", "commitUser": "u",)" + R"( "commitIdentifier": 1, "commitKind": "APPEND", "timeMillis": 100,)" + R"( "totalRecordCount": 10, "deltaRecordCount": 1}, "unknown": 1})"; + ASSERT_OK_AND_ASSIGN(CommitLockResponse granted, + CommitLockResponse::FromJsonString(granted_json)); + ASSERT_TRUE(granted.IsAcquired()); + ASSERT_EQ(std::optional("commit-user"), granted.GetCommitUser()); + ASSERT_EQ(1700000060000, granted.GetExpiresAtMillis()); + ASSERT_EQ(60000, granted.GetLeaseMillis()); + ASSERT_TRUE(granted.GetSnapshot()); + ASSERT_EQ(7, granted.GetSnapshot()->Id()); + + ASSERT_OK_AND_ASSIGN(std::string written, granted.ToJsonString()); + ASSERT_OK_AND_ASSIGN(CommitLockResponse round_tripped, + CommitLockResponse::FromJsonString(written)); + ASSERT_TRUE(round_tripped.IsAcquired()); + ASSERT_EQ(granted.GetCommitUser(), round_tripped.GetCommitUser()); + ASSERT_EQ(granted.GetExpiresAtMillis(), round_tripped.GetExpiresAtMillis()); + ASSERT_EQ(granted.GetLeaseMillis(), round_tripped.GetLeaseMillis()); + ASSERT_EQ(granted.GetSnapshot(), round_tripped.GetSnapshot()); + + ASSERT_OK_AND_ASSIGN( + CommitLockResponse contended, + CommitLockResponse::FromJsonString( + R"({"acquired": false, "commitUser": null, "expiresAtMillis": 0, "leaseMillis": 0,)" + R"( "snapshot": null})")); + ASSERT_FALSE(contended.IsAcquired()); + ASSERT_FALSE(contended.GetCommitUser()); + ASSERT_EQ(0, contended.GetLeaseMillis()); + ASSERT_FALSE(contended.GetSnapshot()); + ASSERT_OK_AND_ASSIGN(std::string contended_json, contended.ToJsonString()); + ASSERT_OK_AND_ASSIGN(CommitLockResponse contended_round_trip, + CommitLockResponse::FromJsonString(contended_json)); + ASSERT_FALSE(contended_round_trip.IsAcquired()); + ASSERT_FALSE(contended_round_trip.GetCommitUser()); + ASSERT_FALSE(contended_round_trip.GetSnapshot()); + + ASSERT_NOK_WITH_MSG(CommitLockResponse::FromJsonString("{}"), "key must exist"); + ASSERT_NOK_WITH_MSG(CommitLockResponse::FromJsonString(R"({"acquired": null})"), + "key must exist"); + ASSERT_NOK_WITH_MSG(CommitLockResponse::FromJsonString(R"({"acquired": "true"})"), + "value must be bool"); + ASSERT_NOK_WITH_MSG(CommitLockResponse::FromJsonString("[]"), "value must be an object"); +} + TEST(RestMessagesTest, CreateDatabaseRequestRoundTrip) { CreateDatabaseRequest request("db1", {{"k1", "v1"}}); ASSERT_OK_AND_ASSIGN(std::string json, request.ToJsonString()); diff --git a/src/paimon/testing/mock/mock_catalog.h b/src/paimon/testing/mock/mock_catalog.h index 67449abf2..1a2f0b7d7 100644 --- a/src/paimon/testing/mock/mock_catalog.h +++ b/src/paimon/testing/mock/mock_catalog.h @@ -31,6 +31,7 @@ #include "paimon/catalog/catalog.h" #include "paimon/catalog/identifier.h" #include "paimon/catalog/table.h" +#include "paimon/core/catalog/commit_lock.h" #include "paimon/core/catalog/version_managed_catalog.h" #include "paimon/core/partition/partition_statistics.h" #include "paimon/core/snapshot.h" @@ -93,6 +94,22 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog { lost_answer_ = status; } + using CommitLockFactory = std::function>( + const Identifier&, const std::optional&, const std::string&)>; + + void SetCommitLockFactory(CommitLockFactory factory) { + commit_lock_factory_ = std::move(factory); + } + + Result> CreateCommitLock( + const Identifier& identifier, const std::optional& table_uuid, + const std::string& commit_user, const CoreOptions& options) override { + if (!commit_lock_factory_) { + return std::unique_ptr(); + } + return commit_lock_factory_(identifier, table_uuid, commit_user); + } + Result> GetTable(const Identifier& identifier) const override { ++get_table_calls_; // Serves the per-table file system when one was set, matching `GetTableFileSystem` @@ -294,6 +311,7 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog { Status load_snapshot_status_; std::vector commit_calls_; std::map options_; + CommitLockFactory commit_lock_factory_; }; } // namespace paimon::test