lxy-9602 commented on code in PR #357:
URL: https://github.com/apache/paimon-cpp/pull/357#discussion_r4032572535
##########
docs/source/user_guide/catalog.rst:
##########
@@ -224,7 +222,73 @@ restored by ``RollbackToAsLatest``. Serialize rollback and
expiration through
the upstream coordinator: a rollback must finish before expiration starts,
so its restored file references are visible to the expiration operation.
+.. warning::
+
+ Expiration reads the retained snapshots of the branch it runs on and of no
+ other, while data files are shared by every branch of the table, so a file
+ that only another branch still refers to is not preserved by that reference
+ and is deleted. This holds for the main branch as much as for the others:
+ expiring the main branch deletes a file only a branch refers to. Expire only
+ where the retained snapshots of the branch cover every file the other
branches
+ still read, or keep those files reachable from the expiring branch through
the
+ upstream coordinator.
+
Review Comment:
Could we add some safeguards here? For example, perhaps expiration should be
unsupported for tables that have branches.
##########
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:
I’d like to understand whether, in the current production code,
`snapshot_manager_` can actually belong to a different branch than the input
`branch`. If that is not possible, could we make this check stricter and fail
explicitly here?
##########
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:
`ASSERT_NOK_WITH_MSG`
##########
test/inte/pk_compaction_inte_test.cpp:
##########
@@ -1954,35 +1954,77 @@ TEST_F(PkCompactionInteTest, WriteAndCompactWithBranch)
{
ASSERT_EQ(compact_after[0]->level, 5);
ASSERT_EQ(compact_after[0]->row_count, 2);
- // Step 4: Fake a DataSplit from compact_after to read and verify the
compacted data.
- {
- ReadContextBuilder read_context_builder(table_path);
- read_context_builder.WithBranch("rt");
- ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish());
- ASSERT_OK_AND_ASSIGN(auto table_read,
TableRead::Create(std::move(read_context)));
+ // Step 4: Commit to branch-rt.
+ CommitContextBuilder commit_builder(table_path, "commit_user_1");
+ commit_builder.AddOption(Options::FILE_SYSTEM,
"local").AddOption(Options::BRANCH, "rt");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> commit_context,
commit_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> file_store_commit,
+ FileStoreCommit::Create(std::move(commit_context)));
+ ASSERT_OK(file_store_commit->Commit(commit_msgs));
- // Build a fake DataSplit using the compact_after file metadata.
- auto fake_split = BuildSplit(commit_msgs, /*snapshot_id=*/1);
- ASSERT_OK_AND_ASSIGN(auto batch_reader,
-
table_read->CreateReader(std::shared_ptr<Split>(fake_split)));
- ASSERT_OK_AND_ASSIGN(auto result_array,
-
ReadResultCollector::CollectResult(std::move(batch_reader)));
+ auto fs = std::make_shared<LocalFileSystem>();
+ ASSERT_OK_AND_ASSIGN(
+ bool branch_append_snapshot,
+ fs->Exists(PathUtil::JoinPath(table_path,
"branch/branch-rt/snapshot/snapshot-2")));
+ ASSERT_TRUE(branch_append_snapshot);
+ ASSERT_OK_AND_ASSIGN(
+ bool branch_compact_snapshot,
+ fs->Exists(PathUtil::JoinPath(table_path,
"branch/branch-rt/snapshot/snapshot-3")));
+ ASSERT_TRUE(branch_compact_snapshot);
+ ASSERT_OK_AND_ASSIGN(bool main_snapshot_exists,
+ fs->Exists(PathUtil::JoinPath(table_path,
"snapshot/snapshot-2")));
+ ASSERT_FALSE(main_snapshot_exists);
+
+ // Step 5: Scan branch-rt and read the compacted file.
+ std::map<std::string, std::string> branch_options =
{{Options::FILE_SYSTEM, "local"},
+ {Options::BRANCH,
"rt"}};
+ ScanContextBuilder scan_context_builder(table_path);
+ scan_context_builder.WithStreamingMode(false)
+ .SetOptions(branch_options)
+ .AddOption(Options::SCAN_MODE, StartupMode::LatestFull().ToString());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context,
scan_context_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan,
+ TableScan::Create(std::move(scan_context)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan());
- arrow::FieldVector fields_with_row_kind = fields;
- fields_with_row_kind.insert(fields_with_row_kind.begin(),
- arrow::field("_VALUE_KIND",
arrow::int8()));
- auto result_type = arrow::struct_(fields_with_row_kind);
+ std::vector<std::shared_ptr<Split>> compacted_splits;
+ for (const std::shared_ptr<Split>& split : plan->Splits()) {
+ auto* split_impl = dynamic_cast<DataSplitImpl*>(split.get());
+ ASSERT_NE(split_impl, nullptr);
+ for (const std::shared_ptr<DataFileMeta>& file :
split_impl->DataFiles()) {
+ if (file->file_name == compact_after[0]->file_name) {
+ ASSERT_EQ(split_impl->SnapshotId(), 3);
+ compacted_splits.push_back(split);
+ break;
+ }
+ }
+ }
+ ASSERT_EQ(compacted_splits.size(), 1u);
Review Comment:
Reconstructing the split inside the integration test feels a bit unusual.
Would it make more sense to just verify it normally with
`ASSERT_EQ(split_impl->SnapshotId(), 3);`, and then use the full split for
reading?
--
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]
