SteNicholas commented on code in PR #357:
URL: https://github.com/apache/paimon-cpp/pull/357#discussion_r4033767190
##########
src/paimon/core/catalog/renaming_snapshot_commit.h:
##########
@@ -47,10 +48,17 @@ class RenamingSnapshotCommit : public SnapshotCommit {
/// @note The atomic rename detects conflicts by snapshot ID, so
`base_snapshot_uuid` is unused.
Result<bool> Commit(const std::optional<std::string>& base_snapshot_uuid,
- const Snapshot& snapshot,
+ const Snapshot& snapshot, const std::string& branch,
const std::vector<PartitionStatistics>& statistics)
override {
PAIMON_ASSIGN_OR_RAISE(std::string json_str, snapshot.ToJsonString());
- std::string snapshot_path =
snapshot_manager_->SnapshotPath(snapshot.Id());
+ // The snapshot directory belongs to a branch, so a commit aimed at
another one writes
+ // through a manager of that branch.
+ std::shared_ptr<SnapshotManager> snapshot_manager = snapshot_manager_;
+ if (BranchManager::NormalizeBranch(branch) !=
snapshot_manager_->Branch()) {
+ snapshot_manager = std::make_shared<SnapshotManager>(
+ snapshot_manager_->Fs(), snapshot_manager_->RootPath(),
branch);
Review Comment:
It cannot in the current production code: `FileStoreCommit::Create()` builds
one branch-aware `SnapshotManager` and hands it both to
`RenamingSnapshotCommit` and to `FileStoreCommitImpl`, which passes
`snapshot_manager_->Branch()` to `SnapshotCommit::Commit()`.
So a mismatch can only be a caller bug, and in 0af0eea it returns `Invalid`
instead of switching to a manager of the other branch.
`RenamingSnapshotCommitTest.TestBranchHasToMatchSnapshotManager` covers both
directions and checks that nothing is written. This is stricter than Java's
`RenamingSnapshotCommit`, which switches with `copyWithBranch`.
##########
src/paimon/core/operation/file_store_commit_impl_test.cpp:
##########
@@ -721,6 +722,194 @@ TEST_F(FileStoreCommitImplTest,
TestCatalogCommitTakesTheSnapshot) {
ASSERT_GT(CountFiles(PathUtil::JoinPath(table_path_, "manifest")), 0u);
}
+TEST_F(FileStoreCommitImplTest, TestCommitToBranch) {
+ CommitContextBuilder missing_builder(table_path_, "commit_user_1");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> missing_context,
+ missing_builder.AddOption(Options::FILE_SYSTEM,
"local")
+ .AddOption(Options::BRANCH, "dev")
+ .Finish());
+ ASSERT_NOK_WITH_MSG(FileStoreCommit::Create(std::move(missing_context)),
+ "not found latest schema in branch dev");
+
+ SchemaManager branch_schema_manager(file_system_, table_path_, "dev");
+ ASSERT_OK(branch_schema_manager.CreateTable(arrow::schema(fields_),
/*partition_keys=*/{"f1"},
+ /*primary_keys=*/{},
/*options=*/{}));
+
+ CommitContextBuilder builder(table_path_, "commit_user_1");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> context,
+ builder.AddOption(Options::FILE_SYSTEM, "local")
+ .AddOption(Options::BRANCH, "dev")
+ .Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit,
+ FileStoreCommit::Create(std::move(context)));
+ ASSERT_OK(commit->Commit(CommitMessagesOfRound(1),
/*commit_identifier=*/1));
+
+ std::string branch_snapshots = PathUtil::JoinPath(table_path_,
"branch/branch-dev/snapshot");
+ ASSERT_OK_AND_ASSIGN(bool branch_exist,
+
file_system_->Exists(PathUtil::JoinPath(branch_snapshots, "snapshot-1")));
+ ASSERT_TRUE(branch_exist);
+ ASSERT_OK_AND_ASSIGN(bool hint_exist,
+
file_system_->Exists(PathUtil::JoinPath(branch_snapshots, "LATEST")));
+ ASSERT_TRUE(hint_exist);
+ ASSERT_OK_AND_ASSIGN(bool main_exist,
file_system_->Exists(PathUtil::JoinPath(
+ table_path_,
"snapshot/snapshot-1")));
+ ASSERT_FALSE(main_exist);
+ ASSERT_GT(CountFiles(PathUtil::JoinPath(table_path_, "manifest")), 0u);
+
+ SnapshotManager branch_snapshot_manager(file_system_, table_path_, "dev");
+ ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> latest,
branch_snapshot_manager.LatestSnapshot());
+ ASSERT_TRUE(latest);
+ ASSERT_EQ(latest.value().Id(), 1);
+}
+
+TEST_F(FileStoreCommitImplTest, TestRequestOnlyCommitToBranch) {
+ SchemaManager branch_schema_manager(file_system_, table_path_, "dev");
+ ASSERT_OK(branch_schema_manager.CreateTable(arrow::schema(fields_),
/*partition_keys=*/{"f1"},
+ /*primary_keys=*/{},
/*options=*/{}));
+
+ CommitContextBuilder builder(table_path_, "commit_user_1");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> context,
+ builder.AddOption(Options::FILE_SYSTEM, "local")
+ .AddOption(Options::BRANCH, "dev")
+ .UseRESTCatalogCommit(true)
+ .WithTableId("table-uuid")
+ .Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit,
+ FileStoreCommit::Create(std::move(context)));
+ ASSERT_OK(commit->Commit(CommitMessagesOfRound(1),
/*commit_identifier=*/1));
+
+ ASSERT_OK_AND_ASSIGN(std::string request_str,
commit->GetLastCommitTableRequest());
+ ASSERT_OK_AND_ASSIGN(CommitTableRequest request,
+ CommitTableRequest::FromJsonString(request_str));
+ ASSERT_EQ(request.GetTableId(), std::optional<std::string>("table-uuid"));
+ ASSERT_EQ(request.GetSnapshot().Id(), 1);
+ ASSERT_EQ(request.GetSnapshot().SchemaId(), 0);
+
+ ASSERT_OK_AND_ASSIGN(bool branch_exist,
+ file_system_->Exists(PathUtil::JoinPath(
+ table_path_,
"branch/branch-dev/snapshot/snapshot-1")));
+ ASSERT_FALSE(branch_exist);
+ ASSERT_OK_AND_ASSIGN(bool main_exist,
file_system_->Exists(PathUtil::JoinPath(
+ table_path_,
"snapshot/snapshot-1")));
+ ASSERT_FALSE(main_exist);
+ ASSERT_GT(CountFiles(PathUtil::JoinPath(table_path_, "manifest")), 0u);
+}
+
+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"},
+ /*primary_keys=*/{},
/*options=*/{}));
+
+ CommitContextBuilder base_builder(table_path_, "commit_user_1");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> base_context,
+ base_builder.AddOption(Options::FILE_SYSTEM, "local")
+ .AddOption(Options::BRANCH, "dev")
+ .Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> base_commit,
+ FileStoreCommit::Create(std::move(base_context)));
+ ASSERT_OK(base_commit->Commit(CommitMessagesOfRound(1),
/*commit_identifier=*/1));
+ SnapshotManager branch_snapshot_manager(file_system_, table_path_, "dev");
+ ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> base_snapshot,
+ branch_snapshot_manager.LatestSnapshot());
+ ASSERT_TRUE(base_snapshot);
+ ASSERT_EQ(base_snapshot.value().Id(), 1);
+ ASSERT_TRUE(base_snapshot.value().Uuid().has_value());
+
+ auto catalog = CreateCatalog();
+ catalog->SetHeldSnapshot(base_snapshot.value());
+ 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<FileStoreCommit> commit,
CreateCatalogCommit(spec));
+ ASSERT_OK(commit->Commit(CommitMessagesOfRound(2),
/*commit_identifier=*/2));
+
+ ASSERT_GT(catalog->LoadSnapshotCalls(), 0u);
+ for (const Identifier& asked : catalog->LoadSnapshotIdentifiers()) {
+ ASSERT_EQ(asked, Identifier("db", "tbl$branch_dev"));
+ }
+ ASSERT_EQ(catalog->CommitCalls().size(), 1u);
+ const MockVersionManagedCatalog::CommitCall& call =
catalog->CommitCalls().front();
+ ASSERT_EQ(call.identifier, Identifier("db", "tbl$branch_dev"));
+ ASSERT_EQ(call.base_snapshot_uuid, base_snapshot.value().Uuid());
+ ASSERT_EQ(call.snapshot.Id(), 2);
+
+ ASSERT_OK_AND_ASSIGN(bool base_published,
+ file_system_->Exists(PathUtil::JoinPath(
+ table_path_,
"branch/branch-dev/snapshot/snapshot-1")));
+ ASSERT_TRUE(base_published);
+ ASSERT_OK_AND_ASSIGN(bool branch_exist,
+ file_system_->Exists(PathUtil::JoinPath(
+ table_path_,
"branch/branch-dev/snapshot/snapshot-2")));
+ ASSERT_FALSE(branch_exist);
+}
+
+TEST_F(FileStoreCommitImplTest, TestBranchCommitReadsTheSchemaFromTheBranch) {
+ SchemaManager branch_schema_manager(file_system_, table_path_, "dev");
+ ASSERT_OK(branch_schema_manager.CreateTable(arrow::schema(fields_),
/*partition_keys=*/{"f1"},
+ /*primary_keys=*/{},
/*options=*/{}));
+
+ std::shared_ptr<MockVersionManagedCatalog> catalog =
CreateCatalogServingSchemaId(7);
+ 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<FileStoreCommit> commit,
CreateCatalogCommit(spec));
+ ASSERT_OK(commit->Commit(CommitMessagesOfRound(1),
/*commit_identifier=*/1));
+
+ ASSERT_EQ(catalog->CommitCalls().size(), 1u);
+ ASSERT_EQ(catalog->CommitCalls().front().identifier, Identifier("db",
"tbl$branch_dev"));
+ ASSERT_EQ(catalog->CommitCalls().front().snapshot.SchemaId(), 0);
+ ASSERT_EQ(catalog->LoadTableSchemaCalls(), 0u);
+
+ catalog->SetTableSchema(nullptr);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> next_commit,
CreateCatalogCommit(spec));
+ ASSERT_OK(next_commit->Commit(CommitMessagesOfRound(2),
/*commit_identifier=*/2));
+
+ ASSERT_EQ(catalog->CommitCalls().size(), 2u);
+ ASSERT_EQ(catalog->CommitCalls().back().snapshot.SchemaId(), 0);
+ ASSERT_EQ(catalog->LoadTableSchemaCalls(), 0u);
+}
+
+TEST_F(FileStoreCommitImplTest, TestBranchCommitNamesTheBranchWhenItFails) {
+ SchemaManager branch_schema_manager(file_system_, table_path_, "dev");
+ ASSERT_OK(branch_schema_manager.CreateTable(arrow::schema(fields_),
/*partition_keys=*/{"f1"},
+ /*primary_keys=*/{},
/*options=*/{}));
+
+ auto catalog = CreateCatalog(Status::IOError("catalog unreachable"));
+ CatalogCommitSpec spec;
+ spec.catalog = catalog;
+ spec.identifier.emplace("db", "tbl$branch_dev");
+ spec.table_id = "table-uuid";
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit,
CreateCatalogCommit(spec));
+
+ Status failed = commit->Commit(CommitMessagesOfRound(1));
+ ASSERT_NOK_WITH_MSG(failed, "catalog unreachable");
+ ASSERT_NE(failed.ToString().find("(branch dev, through catalog"),
std::string::npos)
+ << failed.ToString();
Review Comment:
Done in 0af0eea: both checks use `ASSERT_NOK_WITH_MSG` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
