This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new e209445f fix(commit): handle ambiguous atomic snapshot writes (#397)
e209445f is described below
commit e209445f0e4ec21343ed41349b6531e4f8c12038
Author: lxy <[email protected]>
AuthorDate: Tue Sep 29 17:18:00 2026 +0800
fix(commit): handle ambiguous atomic snapshot writes (#397)
---
src/paimon/core/catalog/renaming_snapshot_commit.h | 23 ++++---
.../core/catalog/renaming_snapshot_commit_test.cpp | 78 ++++++++++++++++++++++
.../core/operation/file_store_commit_impl_test.cpp | 2 +-
3 files changed, 94 insertions(+), 9 deletions(-)
diff --git a/src/paimon/core/catalog/renaming_snapshot_commit.h
b/src/paimon/core/catalog/renaming_snapshot_commit.h
index b70ef0a8..da0269cb 100644
--- a/src/paimon/core/catalog/renaming_snapshot_commit.h
+++ b/src/paimon/core/catalog/renaming_snapshot_commit.h
@@ -65,14 +65,21 @@ class RenamingSnapshotCommit : public SnapshotCommit {
if (is_exist) {
return false;
}
- // To prevent the case where an atomic write times out but actually
succeeds,
- // retrying the commit could lead to the snapshot file being committed
multiple times.
- // Therefore, retries should be handled by the upper layer,
- // which should call FilterAndCommit to avoid duplicate commits.
- // Therefore, we should not trigger cleanup here,
- // as it may delete meta files from a snapshot that was just written
by ourselves,
- // leading to an incomplete or corrupted snapshot.
- PAIMON_RETURN_NOT_OK(fs_->AtomicStore(snapshot_path, json_str));
+ Status commit_status = fs_->AtomicStore(snapshot_path, json_str);
+ if (!commit_status.ok()) {
+ PAIMON_ASSIGN_OR_RAISE(bool snapshot_exists,
fs_->Exists(snapshot_path));
+ if (!snapshot_exists) {
+ return commit_status;
+ }
+
+ std::string committed_json;
+ PAIMON_RETURN_NOT_OK(fs_->ReadFile(snapshot_path,
&committed_json));
+ PAIMON_ASSIGN_OR_RAISE(Snapshot committed_snapshot,
+ Snapshot::FromJsonString(committed_json));
+ if (!(committed_snapshot == snapshot)) {
+ return false;
+ }
+ }
PAIMON_RETURN_NOT_OK(snapshot_manager_->CommitLatestHint(snapshot.Id()));
return true;
}
diff --git a/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp
b/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp
index 4f363d74..4197cb5e 100644
--- a/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp
+++ b/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp
@@ -21,6 +21,7 @@
#include <optional>
#include <string>
+#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/core/partition/partition_statistics.h"
@@ -30,6 +31,15 @@
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
+namespace {
+
+class AtomicStoreMockFileSystem : public LocalFileSystem {
+ public:
+ MOCK_METHOD(Status, AtomicStore, (const std::string& path, const
std::string& content),
+ (override));
+};
+
+} // namespace
TEST(RenamingSnapshotCommitTest, TestSimple) {
auto fs = std::make_shared<LocalFileSystem>();
@@ -106,4 +116,72 @@ TEST(RenamingSnapshotCommitTest,
TestBranchHasToMatchSnapshotManager) {
ASSERT_TRUE(main_success);
}
+TEST(RenamingSnapshotCommitTest,
TestAtomicStoreErrorWithMatchingSnapshotIsSuccess) {
+ auto fs = std::make_shared<AtomicStoreMockFileSystem>();
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto snapshot_manager = std::make_shared<SnapshotManager>(fs, dir->Str());
+ auto commit = std::make_shared<RenamingSnapshotCommit>(fs,
snapshot_manager);
+ Snapshot snapshot = BuildTestSnapshot(1);
+ AtomicStoreMockFileSystem* fs_ptr = fs.get();
+
+ EXPECT_CALL(*fs,
+ AtomicStore(testing::A<const std::string&>(), testing::A<const
std::string&>()))
+ .WillOnce(testing::Invoke(
+ [fs_ptr](const std::string& path, const std::string& content) ->
Status {
+ PAIMON_RETURN_NOT_OK(fs_ptr->FileSystem::AtomicStore(path,
content));
+ return Status::IOError("response lost after atomic store");
+ }));
+
+ ASSERT_OK_AND_ASSIGN(bool success, commit->Commit(std::nullopt, snapshot,
+
BranchManager::DEFAULT_MAIN_BRANCH, {}));
+ ASSERT_TRUE(success);
+ std::string latest_hint;
+ ASSERT_OK(fs->ReadFile(PathUtil::JoinPath(dir->Str(), "snapshot/LATEST"),
&latest_hint));
+ ASSERT_EQ("1", latest_hint);
+}
+
+TEST(RenamingSnapshotCommitTest,
TestAtomicStoreErrorWithConflictingSnapshotReturnsFalse) {
+ auto fs = std::make_shared<AtomicStoreMockFileSystem>();
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto snapshot_manager = std::make_shared<SnapshotManager>(fs, dir->Str());
+ auto commit = std::make_shared<RenamingSnapshotCommit>(fs,
snapshot_manager);
+ Snapshot snapshot = BuildTestSnapshot(1);
+ ASSERT_OK_AND_ASSIGN(std::string conflicting_json,
BuildTestSnapshot(2).ToJsonString());
+ AtomicStoreMockFileSystem* fs_ptr = fs.get();
+
+ EXPECT_CALL(*fs,
+ AtomicStore(testing::A<const std::string&>(), testing::A<const
std::string&>()))
+ .WillOnce(testing::Invoke(
+ [fs_ptr, &conflicting_json](const std::string& path, const
std::string&) -> Status {
+ PAIMON_RETURN_NOT_OK(fs_ptr->FileSystem::AtomicStore(path,
conflicting_json));
+ return Status::Exist("another writer committed this snapshot
id");
+ }));
+
+ ASSERT_OK_AND_ASSIGN(bool success, commit->Commit(std::nullopt, snapshot,
+
BranchManager::DEFAULT_MAIN_BRANCH, {}));
+ ASSERT_FALSE(success);
+ ASSERT_OK_AND_ASSIGN(bool hint_exists,
+ fs->Exists(PathUtil::JoinPath(dir->Str(),
"snapshot/LATEST")));
+ ASSERT_FALSE(hint_exists);
+}
+
+TEST(RenamingSnapshotCommitTest,
TestAtomicStoreErrorWithMissingSnapshotReturnsError) {
+ auto fs = std::make_shared<AtomicStoreMockFileSystem>();
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto snapshot_manager = std::make_shared<SnapshotManager>(fs, dir->Str());
+ auto commit = std::make_shared<RenamingSnapshotCommit>(fs,
snapshot_manager);
+ Snapshot snapshot = BuildTestSnapshot(1);
+
+ EXPECT_CALL(*fs,
+ AtomicStore(testing::A<const std::string&>(), testing::A<const
std::string&>()))
+ .WillOnce(testing::Return(Status::IOError("atomic store failed")));
+
+ ASSERT_NOK_WITH_MSG(
+ commit->Commit(std::nullopt, snapshot,
BranchManager::DEFAULT_MAIN_BRANCH, {}),
+ "atomic store failed");
+}
+
} // namespace paimon::test
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 b5d2b760..02ce10c4 100644
--- a/src/paimon/core/operation/file_store_commit_impl_test.cpp
+++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp
@@ -1890,7 +1890,7 @@ TEST_F(FileStoreCommitImplTest,
TestCommitWithAtomicWriteSnapshotTimeoutAndActua
"/parquet/append_09.db/append_09/commit_messages/commit_messages-01",
/*version=*/3);
ASSERT_GT(msgs.size(), 0);
- ASSERT_NOK(commit->Commit(msgs, /*commit_identifier=*/1));
+ ASSERT_OK(commit->Commit(msgs, /*commit_identifier=*/1));
ASSERT_OK_AND_ASSIGN(
bool exist, file_system_->Exists(PathUtil::JoinPath(table_path,
"snapshot/snapshot-6")));
ASSERT_TRUE(exist);
