SteNicholas commented on code in PR #357:
URL: https://github.com/apache/paimon-cpp/pull/357#discussion_r4033766730
##########
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:
Done in eb06e6e. The test no longer picks the split holding the compacted
file: it asserts `SnapshotId() == 3` for every split of `branch-rt` and reads
all of them, so the expected result is now every row of the branch.
##########
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:
Done in eb06e6e. `Expire()` now returns `NotImplemented` before reading any
snapshot or deleting any file when it runs on a branch other than main, or when
it runs on main and finds another branch under `branch/branch-<name>` of the
table path; that error lists the branches found.
A branch held only by a catalog, or one created while expiration runs, is
not found by this check, so `FileStoreCommit::Expire()` and the catalog and
clean guides call that out and ask to serialize branch creation and expiration.
`TestExpireOnBranchIsNotSupported` and
`TestExpireOnMainOfTableWithBranchesIsNotSupported` cover each refusal and
check that no snapshot, manifest or data file is deleted.
##########
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 eb06e6e 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 eb06e6e: 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]
