diff --git a/docs/source/user_guide/catalog.rst b/docs/source/user_guide/catalog.rst index b843b5086..428839059 100644 --- a/docs/source/user_guide/catalog.rst +++ b/docs/source/user_guide/catalog.rst @@ -180,11 +180,9 @@ Its metadata files are retained. Use ``FilterAndCommit`` for batch recovery; after a real-time error, recreate the writer and context and recover from durable offsets as described by ``CommitWithProgress``. -Only the main branch is supported by ``WithCatalog``. Both builders reject other -branches in the identifier or ``branch`` option; the write builder also checks -``WithBranch``. Explicit ``tbl$branch_main`` and ``branch=main`` are accepted. - -The current schema and latest snapshot need no files under the table path. +The main branch's current schema and latest snapshot need no files under the table +path. A branch needs its schema published under its own directory, while its +latest snapshot can be held by the catalog alone, as described below. A catalog response of ``{"snapshot": null}`` means the table has no snapshot. Only a catalog reporting that snapshot loading is unsupported falls back to file-system lookup; other catalog errors propagate to the caller. @@ -193,7 +191,8 @@ Read ``Table::CatalogUuid()`` before preparing changes and pass it to ``WithTableId`` so the server can reject commits to a dropped and recreated table. ``Table::Uuid()`` can fall back to the table name and must not supply this id. An unset table id is sent as null and subject to server validation. Each commit -attempt reloads the catalog's current schema id for the new snapshot. +attempt to the main branch reloads the catalog's current schema id for the new +snapshot; on a branch it reloads the id published under the branch. .. note:: @@ -224,7 +223,76 @@ 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:: + + ``Expire`` returns ``NotImplemented`` before reading any snapshot or deleting + any file when it runs on a branch other than main, or finds one under + ``branch/branch-`` of the table path. Data files are shared by every + branch of the table, while expiration reads only the retained snapshots of the + branch it runs on, so it could delete a file another branch still refers to. + A branch the catalog holds without that directory, or one created while + expiration runs, is not found: do not expire a table with such a branch, and + serialize branch creation and expiration through the upstream coordinator. + The C++ REST catalog covers the database, table, snapshot and commit operations of the ``Catalog`` API. The parts of the Java REST catalog that have no C++ counterpart yet — altering a database or a table, views, functions, partitions, -tags, branch management and consumers — are not supported. +tags, branch management and consumers — are not supported. Creating, deleting +and merging a branch is what branch management covers; committing to a branch +that already exists is supported, as described below. + +Committing to a branch +~~~~~~~~~~~~~~~~~~~~~~ +This library creates no branch: one has to exist, with its schema published under +``branch/branch-``, before a write or a commit can be aimed at it. + +Naming the branch +^^^^^^^^^^^^^^^^^ +A branch is addressed by the identifier of ``WithCatalog``, which +``paimon::Identifier("db", "tbl", "dev")`` builds as ``tbl$branch_dev``, so that +the catalog answers for the branch: the snapshot it loads is the branch's, and +the commit request is addressed to that branch by that identifier rather than by +anything in its body. The ``branch`` option and, for writers, ``WithBranch`` may +name the same branch as well; naming two different branches is rejected rather +than resolved, and ``tbl$branch_main``, ``branch=main`` and an empty name all +mean the main branch. Naming a branch only in the option or ``WithBranch`` while +the identifier names the bare table is rejected, as the catalog would then answer +for the table while the files were written for the branch. + +Names a catalog cannot tell apart from another object are rejected where a +builder sees them: a name differing from ``main`` by case alone, such as +``tbl$branch_MAIN`` or ``branch=MAIN``, which a catalog reads as the main branch +while its snapshots go to a directory of their own; and a name holding a ``$``, +such as ``branch=dev$options``, which ``tbl$branch_dev$options`` reads back as +the ``options`` system table of branch ``dev``. ``paimon::Identifier("db", +"tbl", "MAIN")`` never reaches that check: the constructor folds every spelling +of ``main`` into the bare table name, as the Java client does, and so names the +main branch. + +What a branch keeps and what it shares +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +A branch other than the main one publishes its schema and real-time offsets +under ``branch/branch-``, where a snapshot written to the file system goes +as well; the main branch keeps these files under the table path. A +version-managed catalog can hold the latest snapshot of either branch alone. +Data and manifests are shared by all branches under the table path, +which is why expiration is not supported on a table with branches, as noted +above. + +A write or a commit aimed at a branch reads that schema rather than the +catalog's, just as a read of that branch does, and asks the catalog for no schema +at all, so a catalog serving only the table's own current schema still takes the +branch's snapshots. The new snapshot records the schema id published there as +well, which is the id a read of the branch resolves under +``branch/branch-``; only on the main branch is the recorded id the one the +catalog reports, as its current schema may live nowhere else. Data files record +the schema they were written with, as they do on the main branch. + +Addressing a branch without a catalog client +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +``UseRESTCatalogCommit`` builds the request instead of sending it, and takes the +branch from the ``branch`` option alone, as it is configured without an +identifier. The request body carries the table id but no table name and no +branch, so the branch has to appear in the URL the caller sends that body to: +the commit endpoint of ``tbl$branch_dev``, not the one of ``tbl``. A body sent +to the bare table's URL publishes the branch's snapshot on the main branch. diff --git a/docs/source/user_guide/clean.rst b/docs/source/user_guide/clean.rst index 32099dd86..d0df8186f 100644 --- a/docs/source/user_guide/clean.rst +++ b/docs/source/user_guide/clean.rst @@ -113,6 +113,13 @@ Description - The snapshot expiration interface deletes data files according to the expiration policy and returns the number of snapshots deleted. +- Expiration on a branch other than main, or on a table with such a branch under + ``branch/branch-``, is not supported: branches share the table's data + files, so expiration returns ``NotImplemented`` before deleting anything. +- A branch is found only by that directory. A branch held only by a catalog, or + one created while expiration runs, is not found, and expiration could delete + data files it still refers to: do not expire a table with such a branch, and + serialize branch creation and expiration. Detailed Steps ~~~~~~~~~~~~~~ diff --git a/docs/source/user_guide/write.rst b/docs/source/user_guide/write.rst index 8f4e3e90d..422b109b0 100644 --- a/docs/source/user_guide/write.rst +++ b/docs/source/user_guide/write.rst @@ -172,14 +172,16 @@ Operational Flow For a table of a catalog which owns its versions, such as a REST catalog, it passes that catalog and the table identifier to ``WriteContextBuilder::WithCatalog`` and ``CommitContextBuilder::WithCatalog``. - The writer loads the current schema and latest snapshot from the catalog. - On ``Commit``, the snapshot and - the statistics of the change go to the catalog, which decides whether this - commit wins, and one that lost the race is rebased and retried the way a - file-system commit is. A caller without a catalog client can instead enable - REST catalog commit mode, call ``Commit``, obtain the JSON request from - ``GetLastCommitTableRequest``, and send that request to the REST catalog - itself. + The writer loads the latest snapshot from the catalog, and the current schema + from it as well when it writes to the main branch; a write aimed at another + branch reads the schema published under ``branch/branch-`` instead. On + ``Commit``, the snapshot and the statistics of the change go to the catalog, + which decides whether this commit wins, and one that lost the race is rebased + and retried the way a file-system commit is. A caller without a catalog + client can instead enable REST catalog commit mode, call ``Commit``, obtain + the JSON request from ``GetLastCommitTableRequest``, and send that request to + the REST catalog itself. That request body names no branch, so a commit aimed + at one has to be sent to the URL of ``tbl$branch_dev`` rather than of ``tbl``. 6. The local committer or REST catalog validates the messages, updates manifests/metadata, and finalizes the snapshot atomically. diff --git a/include/paimon/catalog/identifier.h b/include/paimon/catalog/identifier.h index 82ea2e3a6..bcc06208a 100644 --- a/include/paimon/catalog/identifier.h +++ b/include/paimon/catalog/identifier.h @@ -37,6 +37,19 @@ class PAIMON_EXPORT Identifier { explicit Identifier(const std::string& table); Identifier(const std::string& database, const std::string& table); + /// Names `branch` of `table`: every branch but the main one is addressed by the object name + /// this builds, which joins them as `tbl$branch_dev`. + /// + /// The branch is matched against the main branch ignoring case, as the Java client matches it, + /// so `MAIN` names the main branch here rather than a branch of its own. A branch of only + /// whitespace names it too, which the Java client does not do: this library reads a blank + /// branch as the main branch everywhere. + /// + /// @param database Database holding the table. + /// @param table Data table name, carrying no branch or system table suffix. + /// @param branch Branch of `table`; any spelling of `main`, an empty and a blank name all + /// name the main branch. + Identifier(const std::string& database, const std::string& table, const std::string& branch); bool operator==(const Identifier& other) const; const std::string& GetDatabaseName() const; diff --git a/include/paimon/commit_context.h b/include/paimon/commit_context.h index aa2eef266..ffd40ac07 100644 --- a/include/paimon/commit_context.h +++ b/include/paimon/commit_context.h @@ -195,8 +195,11 @@ class PAIMON_EXPORT CommitContextBuilder { /// `Expire()` manages file-system snapshots after confirming that the catalog's current /// snapshot and retained history are published there. Configure writers with the same catalog /// via `WriteContextBuilder::WithCatalog()`. + /// A branch is addressed by the identifier, as `tbl$branch_dev`, so that the catalog answers + /// for that branch. Its schema is read from the branch directory rather than from the catalog, + /// just as a read of that branch reads it. Configure the writer with the same identifier. /// @param catalog Non-null catalog, kept alive by this context. - /// @param identifier Table to commit to. Only the main branch is supported. + /// @param identifier Table to commit to, naming a branch of it as `tbl$branch_dev`. /// @return Reference to this builder for method chaining. CommitContextBuilder& WithCatalog(const std::shared_ptr& catalog, const Identifier& identifier); diff --git a/include/paimon/file_store_commit.h b/include/paimon/file_store_commit.h index d1553c714..776ad52a8 100644 --- a/include/paimon/file_store_commit.h +++ b/include/paimon/file_store_commit.h @@ -164,7 +164,9 @@ class PAIMON_EXPORT FileStoreCommit { const std::vector>& commit_messages, int64_t commit_identifier, std::optional watermark = std::nullopt) = 0; - /// Returns the request from the latest commit attempt, including failed attempts. + /// Returns the request from the latest commit attempt, including failed attempts. Each attempt + /// clears the request of the one before it, so an attempt that failed before building one, as + /// one naming a branch a catalog cannot address does, leaves this returning an error. /// Catalog commits send requests automatically; `UseRESTCatalogCommit()` only prepares them. /// @note Temporary interface for internal use, will be removed in the future. /// @return JSON with `tableId`, `baseSnapshotUuid`, `snapshot` and `statistics`. @@ -179,6 +181,12 @@ class PAIMON_EXPORT FileStoreCommit { /// catalog or retained-metadata read errors propagate without deleting files. /// Retained snapshots with index manifests return `NotImplemented` before deletion. /// @note Coordinate rollback and expiration so they do not run concurrently. + /// @note Returns `NotImplemented` before any snapshot is read or any file is deleted when this + /// commit is on a branch other than main, or the table path holds such a branch under + /// `branch/branch-`: branches share the table's data files, while expiration only + /// reads the retained snapshots of its own branch. A branch held only by a catalog, or + /// created while expiration runs, is not found; do not expire a table with such a + /// branch, and serialize branch creation and expiration. /// @return Result indicating the number of expired items or an error status. virtual Result Expire() = 0; diff --git a/include/paimon/read_context.h b/include/paimon/read_context.h index 25cf7ec5c..306e7350c 100644 --- a/include/paimon/read_context.h +++ b/include/paimon/read_context.h @@ -443,9 +443,12 @@ class PAIMON_EXPORT ReadContextBuilder { /// Paimon supports branching for data versioning and time travel queries. /// This method allows reading from a specific branch instead of the main branch. /// + /// The `branch` option names the branch too, as it does for a scan of one branch; naming two + /// different branches is refused rather than silently resolved. + /// /// @param branch Name of the branch to read from. /// @return Reference to this builder for method chaining. - /// @note Default branch is "main" if not specified. + /// @note Default branch is "main" if not specified. An empty name is the main branch. ReadContextBuilder& WithBranch(const std::string& branch); /// Sets a mapping from URI schemes (e.g., "file", "oss") to registered file system diff --git a/include/paimon/write_context.h b/include/paimon/write_context.h index 1a6e5271d..e0a940eb0 100644 --- a/include/paimon/write_context.h +++ b/include/paimon/write_context.h @@ -238,6 +238,11 @@ class PAIMON_EXPORT WriteContextBuilder { WriteContextBuilder& WithWriteId(int32_t write_id); /// Write to specific branch, default is main. + /// + /// The identifier of `WithCatalog()` and the `branch` option may name the branch too; naming + /// two different ones is refused rather than silently resolved. An empty name is the main + /// branch. + /// @param branch Name of the branch to write to. /// @return Reference to this builder for method chaining. WriteContextBuilder& WithBranch(const std::string& branch); @@ -258,9 +263,12 @@ class PAIMON_EXPORT WriteContextBuilder { /// snapshot. Use the same catalog and identifier as `CommitContextBuilder` to restore and /// refresh real-time progress. /// - /// Only the main branch is supported. Data, manifests and historical metadata remain on the - /// file system. The catalog supplies that file system unless overridden - /// by `WithFileSystem()` or `WithFileSystemSchemeToIdentifierMap()`. + /// A branch is addressed by the identifier, as `tbl$branch_dev`, so that the catalog answers + /// for that branch. Its schema is read from the branch directory rather than from the catalog, + /// just as a read of that branch reads it. + /// Data, manifests and historical metadata remain on the file system. The catalog supplies + /// that file system unless overridden by `WithFileSystem()` or + /// `WithFileSystemSchemeToIdentifierMap()`. /// @param catalog Non-null catalog, kept alive while the writer uses its snapshot loader. /// @param identifier The native table to write to. /// @return Reference to this builder for method chaining. diff --git a/src/paimon/core/catalog/catalog_snapshot_commit.h b/src/paimon/core/catalog/catalog_snapshot_commit.h index 97c3222b2..0c47c99fb 100644 --- a/src/paimon/core/catalog/catalog_snapshot_commit.h +++ b/src/paimon/core/catalog/catalog_snapshot_commit.h @@ -26,9 +26,13 @@ #include "paimon/catalog/catalog.h" #include "paimon/catalog/identifier.h" +#include "paimon/core/catalog/catalog_utils.h" #include "paimon/core/catalog/commit_table_request.h" #include "paimon/core/catalog/snapshot_commit.h" #include "paimon/core/catalog/version_managed_catalog.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/result.h" +#include "paimon/status.h" namespace paimon { @@ -53,8 +57,14 @@ class CatalogSnapshotCommit : public SnapshotCommit { : identifier_("", ""), table_id_(table_id) {} Result Commit(const std::optional& base_snapshot_uuid, - const Snapshot& snapshot, + const Snapshot& snapshot, const std::string& branch, const std::vector& statistics) override { + // Dropped before anything can refuse this attempt, so that what a caller reads back is + // the attempt it just made or nothing at all, never the snapshot of an earlier one. + commit_table_request_.reset(); + // Refused before the request is built: a branch whose object name reads back as another + // object is one that neither this nor the caller of a request can commit to. + PAIMON_RETURN_NOT_OK(BranchManager::CheckCatalogAddressableBranch(branch)); commit_table_request_ = CommitTableRequest(table_id_, base_snapshot_uuid, snapshot, statistics); if (catalog_ == nullptr) { @@ -65,8 +75,11 @@ class CatalogSnapshotCommit : public SnapshotCommit { "this catalog does not manage the versions of its tables, so there is nothing to " "commit the snapshot to"); } - return version_managed_catalog_->CommitSnapshot(identifier_, table_id_, base_snapshot_uuid, - snapshot, statistics); + // The branch this commit is aimed at wins over the one the identifier was built for. + PAIMON_ASSIGN_OR_RAISE(Identifier branch_identifier, + CatalogUtils::BranchIdentifier(identifier_, branch)); + return version_managed_catalog_->CommitSnapshot(branch_identifier, table_id_, + base_snapshot_uuid, snapshot, statistics); } std::string DescribeTarget() const override { @@ -78,12 +91,16 @@ class CatalogSnapshotCommit : public SnapshotCommit { return catalog_ == nullptr; } - /// Returns the request from the latest attempt, including refusals and errors. + /// Returns the request from the latest attempt, including refusals and errors. An attempt + /// refused before the request is built, as one naming an unaddressable branch is, leaves no + /// request behind rather than the one of the attempt before it. Result GetLastCommitTableRequest() override { if (commit_table_request_) { return commit_table_request_.value().ToJsonString(); } else { - return Status::Invalid("Should call Commit first before GetLastCommitTableRequest."); + return Status::Invalid( + "Should call Commit first before GetLastCommitTableRequest. An attempt refused " + "before it built a request leaves none behind."); } } diff --git a/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp b/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp index bf3b2d639..b062c0027 100644 --- a/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp +++ b/src/paimon/core/catalog/catalog_snapshot_commit_test.cpp @@ -47,7 +47,8 @@ TEST(CatalogSnapshotCommitTest, TestCommitThroughCatalog) { std::vector statistics = { PartitionStatistics({{"f1", "20"}}, 1, 541, 1, 1724090888743, -1), PartitionStatistics({{"f1", "10"}}, 4, 1118, 2, 1724090888727, 2)}; - ASSERT_OK_AND_ASSIGN(bool success, commit.Commit("base-snapshot-uuid", snapshot, statistics)); + ASSERT_OK_AND_ASSIGN(bool success, commit.Commit("base-snapshot-uuid", snapshot, + Identifier::kDefaultMainBranch, statistics)); ASSERT_TRUE(success); ASSERT_EQ(catalog->CommitCalls().size(), 1u); @@ -66,7 +67,8 @@ TEST(CatalogSnapshotCommitTest, TestCommitThroughCatalog) { CommitTableRequest("table-uuid", "base-snapshot-uuid", snapshot, statistics)); Snapshot next_snapshot = BuildTestSnapshot(3, "snapshot-uuid-3"); - ASSERT_OK_AND_ASSIGN(success, commit.Commit(snapshot.Uuid(), next_snapshot, {})); + ASSERT_OK_AND_ASSIGN( + success, commit.Commit(snapshot.Uuid(), next_snapshot, Identifier::kDefaultMainBranch, {})); ASSERT_TRUE(success); ASSERT_EQ(catalog->CommitCalls().size(), 2u); const MockVersionManagedCatalog::CommitCall& next_call = catalog->CommitCalls().back(); @@ -81,11 +83,39 @@ TEST(CatalogSnapshotCommitTest, TestCommitThroughCatalog) { ASSERT_EQ(next_request, CommitTableRequest("table-uuid", snapshot.Uuid(), next_snapshot, {})); } +TEST(CatalogSnapshotCommitTest, TestCommitToBranch) { + auto catalog = std::make_shared(true); + CatalogSnapshotCommit commit(catalog, Identifier("db", "tbl"), "table-uuid"); + ASSERT_OK_AND_ASSIGN( + bool success, + commit.Commit(std::nullopt, BuildTestSnapshot(1, "snapshot-uuid-1"), "dev", {})); + ASSERT_TRUE(success); + ASSERT_EQ(catalog->CommitCalls().back().identifier, Identifier("db", "tbl$branch_dev")); + + ASSERT_OK(commit.Commit(std::nullopt, BuildTestSnapshot(2, "snapshot-uuid-2"), + Identifier::kDefaultMainBranch, {})); + ASSERT_EQ(catalog->CommitCalls().back().identifier, Identifier("db", "tbl")); + + ASSERT_NOK_WITH_MSG( + commit.Commit(std::nullopt, BuildTestSnapshot(3, "snapshot-uuid-3"), "MAIN", {}), + "a catalog names branch 'MAIN' as it names the main branch"); + ASSERT_EQ(catalog->CommitCalls().size(), 2u); + + CatalogSnapshotCommit branch_commit(catalog, Identifier("db", "tbl$branch_dev"), "table-uuid"); + ASSERT_OK( + branch_commit.Commit(std::nullopt, BuildTestSnapshot(4, "snapshot-uuid-4"), "release", {})); + ASSERT_EQ(catalog->CommitCalls().back().identifier, Identifier("db", "tbl$branch_release")); + ASSERT_OK(branch_commit.Commit(std::nullopt, BuildTestSnapshot(5, "snapshot-uuid-5"), + Identifier::kDefaultMainBranch, {})); + ASSERT_EQ(catalog->CommitCalls().back().identifier, Identifier("db", "tbl")); +} + TEST(CatalogSnapshotCommitTest, TestLostRaceIsNotAnError) { auto catalog = std::make_shared(false); CatalogSnapshotCommit commit(catalog, Identifier("db", "tbl"), std::nullopt); - ASSERT_OK_AND_ASSIGN( - bool success, commit.Commit("base-snapshot-uuid", BuildTestSnapshot(2, std::nullopt), {})); + ASSERT_OK_AND_ASSIGN(bool success, + commit.Commit("base-snapshot-uuid", BuildTestSnapshot(2, std::nullopt), + Identifier::kDefaultMainBranch, {})); ASSERT_FALSE(success); ASSERT_EQ(catalog->CommitCalls().size(), 1u); ASSERT_EQ(catalog->CommitCalls().front().table_uuid, std::nullopt); @@ -95,7 +125,8 @@ TEST(CatalogSnapshotCommitTest, TestCatalogFailurePropagates) { auto catalog = std::make_shared(Status::IOError("catalog unreachable")); CatalogSnapshotCommit commit(catalog, Identifier("db", "tbl"), "table-uuid"); - ASSERT_NOK_WITH_MSG(commit.Commit(std::nullopt, BuildTestSnapshot(1, std::nullopt), {}), + ASSERT_NOK_WITH_MSG(commit.Commit(std::nullopt, BuildTestSnapshot(1, std::nullopt), + Identifier::kDefaultMainBranch, {}), "catalog unreachable"); ASSERT_OK_AND_ASSIGN(std::string request_str, commit.GetLastCommitTableRequest()); @@ -109,7 +140,8 @@ TEST(CatalogSnapshotCommitTest, TestCatalogWhichDoesNotManageVersionsIsRefused) auto catalog = std::make_shared(); catalog->SetSupportsVersionManagement(false); CatalogSnapshotCommit commit(catalog, Identifier("db", "tbl"), "table-uuid"); - ASSERT_NOK_WITH_MSG(commit.Commit(std::nullopt, BuildTestSnapshot(1, std::nullopt), {}), + ASSERT_NOK_WITH_MSG(commit.Commit(std::nullopt, BuildTestSnapshot(1, std::nullopt), + Identifier::kDefaultMainBranch, {}), "does not manage the versions of its tables"); ASSERT_TRUE(catalog->CommitCalls().empty()); } @@ -122,7 +154,8 @@ TEST(CatalogSnapshotCommitTest, TestBuildRequestWithoutCatalog) { ASSERT_EQ(commit.DescribeTarget(), "commit table request built, not sent"); Snapshot snapshot = BuildTestSnapshot(2, std::nullopt); - ASSERT_OK_AND_ASSIGN(bool success, commit.Commit("base-snapshot-uuid", snapshot, {})); + ASSERT_OK_AND_ASSIGN(bool success, commit.Commit("base-snapshot-uuid", snapshot, + Identifier::kDefaultMainBranch, {})); ASSERT_TRUE(success); ASSERT_OK_AND_ASSIGN(std::string request_str, commit.GetLastCommitTableRequest()); @@ -132,9 +165,24 @@ TEST(CatalogSnapshotCommitTest, TestBuildRequestWithoutCatalog) { ASSERT_EQ(request.GetBaseSnapshotUuid(), std::optional("base-snapshot-uuid")); ASSERT_EQ(request.GetSnapshot().Id(), 2); + ASSERT_OK_AND_ASSIGN( + bool branch_success, + commit.Commit(std::nullopt, BuildTestSnapshot(3, std::nullopt), "dev", {})); + ASSERT_TRUE(branch_success); + ASSERT_OK_AND_ASSIGN(std::string branch_request_str, commit.GetLastCommitTableRequest()); + ASSERT_OK_AND_ASSIGN(CommitTableRequest branch_request, + CommitTableRequest::FromJsonString(branch_request_str)); + ASSERT_EQ(branch_request.GetSnapshot().Id(), 3); + + ASSERT_NOK_WITH_MSG(commit.Commit(std::nullopt, BuildTestSnapshot(4, std::nullopt), "MAIN", {}), + "a catalog names branch 'MAIN' as it names the main branch"); + ASSERT_NOK_WITH_MSG(commit.GetLastCommitTableRequest(), + "Should call Commit first before GetLastCommitTableRequest."); + CatalogSnapshotCommit bare_commit; ASSERT_OK_AND_ASSIGN(bool bare_success, - bare_commit.Commit(std::nullopt, BuildTestSnapshot(1, std::nullopt), {})); + bare_commit.Commit(std::nullopt, BuildTestSnapshot(1, std::nullopt), + Identifier::kDefaultMainBranch, {})); ASSERT_TRUE(bare_success); ASSERT_OK_AND_ASSIGN(std::string bare_request_str, bare_commit.GetLastCommitTableRequest()); ASSERT_OK_AND_ASSIGN(CommitTableRequest bare_request, diff --git a/src/paimon/core/catalog/catalog_utils.cpp b/src/paimon/core/catalog/catalog_utils.cpp index 72f817dcc..1f2a3800a 100644 --- a/src/paimon/core/catalog/catalog_utils.cpp +++ b/src/paimon/core/catalog/catalog_utils.cpp @@ -69,6 +69,12 @@ Status CatalogUtils::CheckNotSystemTable(const Identifier& identifier, const std return Status::OK(); } +Result CatalogUtils::BranchIdentifier(const Identifier& identifier, + const std::string& branch) { + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + return Identifier(identifier.GetDatabaseName(), data_table_name, branch); +} + Status CatalogUtils::CheckNotBranch(const Identifier& identifier, const std::string& action) { PAIMON_ASSIGN_OR_RAISE(std::optional branch, identifier.GetBranchName()); if (branch) { diff --git a/src/paimon/core/catalog/catalog_utils.h b/src/paimon/core/catalog/catalog_utils.h index 4c0ea6d20..565a6aef5 100644 --- a/src/paimon/core/catalog/catalog_utils.h +++ b/src/paimon/core/catalog/catalog_utils.h @@ -50,6 +50,14 @@ class CatalogUtils { /// Fails when `identifier` carries a "$branch_" suffix. static Status CheckNotBranch(const Identifier& identifier, const std::string& action); + /// Returns the identifier a catalog is told `branch` of the data table of `identifier` by: + /// the bare data table for the main branch, and the branch object name, as `tbl$branch_dev`, + /// for any other. + /// Every call a commit or a write makes to the catalog goes to this one identifier, so that + /// what is read and what is committed name one object. + static Result BranchIdentifier(const Identifier& identifier, + const std::string& branch); + /// Fails when `db_name` cannot be used as a single path component, which is required to /// keep the database path under the warehouse. static Status CheckValidDatabaseName(const std::string& db_name); diff --git a/src/paimon/core/catalog/identifier.cpp b/src/paimon/core/catalog/identifier.cpp index 32ba446aa..18fcc7975 100644 --- a/src/paimon/core/catalog/identifier.cpp +++ b/src/paimon/core/catalog/identifier.cpp @@ -35,12 +35,29 @@ const char Identifier::kSystemTableSplitter[] = "$"; const char Identifier::kSystemBranchPrefix[] = "branch_"; const char Identifier::kDefaultMainBranch[] = "main"; +namespace { + +std::string BranchObjectName(const std::string& table, const std::string& branch) { + if (StringUtils::IsNullOrWhitespaceOnly(branch) || + StringUtils::EqualsIgnoreCase(branch, Identifier::kDefaultMainBranch)) { + return table; + } + return fmt::format("{}{}{}{}", table, Identifier::kSystemTableSplitter, + Identifier::kSystemBranchPrefix, branch); +} + +} // namespace + Identifier::Identifier(const std::string& table) : Identifier(std::string(kUnknownDatabase), table) {} Identifier::Identifier(const std::string& database, const std::string& table) : database_(database), table_(table) {} +Identifier::Identifier(const std::string& database, const std::string& table, + const std::string& branch) + : Identifier(database, BranchObjectName(table, branch)) {} + bool Identifier::operator==(const Identifier& other) const { if (this == &other) { return true; @@ -143,7 +160,10 @@ Status Identifier::SplitTableName() const { } else { return Status::Invalid(fmt::format("Invalid table name: {}", table_)); } - if (data_table.empty() || (branch && branch->empty()) || + // A blank branch names the main branch everywhere else here, and the constructor above folds + // it into the bare table name, so an object name spelling one was hand-built and would be + // addressed as a branch of its own. + if (data_table.empty() || (branch && StringUtils::IsNullOrWhitespaceOnly(branch.value())) || (system_table && system_table->empty())) { return Status::Invalid(fmt::format("Invalid table name: {}", table_)); } diff --git a/src/paimon/core/catalog/identifier_test.cpp b/src/paimon/core/catalog/identifier_test.cpp index 50afa6fc7..d358a9255 100644 --- a/src/paimon/core/catalog/identifier_test.cpp +++ b/src/paimon/core/catalog/identifier_test.cpp @@ -99,6 +99,21 @@ TEST(IdentifierTest, ParseBranchTable) { ASSERT_FALSE(is_system_table); } +TEST(IdentifierTest, BranchConstructor) { + Identifier branch_id("db", "tbl", "dev"); + ASSERT_EQ(branch_id.GetTableName(), "tbl$branch_dev"); + ASSERT_OK_AND_ASSIGN(std::string data_table_name, branch_id.GetDataTableName()); + ASSERT_EQ(data_table_name, "tbl"); + ASSERT_OK_AND_ASSIGN(std::optional branch_name, branch_id.GetBranchName()); + ASSERT_TRUE(branch_name); + ASSERT_EQ(branch_name.value(), "dev"); + + ASSERT_EQ(Identifier("db", "tbl", Identifier::kDefaultMainBranch), Identifier("db", "tbl")); + ASSERT_EQ(Identifier("db", "tbl", "MAIN"), Identifier("db", "tbl")); + ASSERT_EQ(Identifier("db", "tbl", ""), Identifier("db", "tbl")); + ASSERT_EQ(Identifier("db", "tbl", " "), Identifier("db", "tbl")); +} + TEST(IdentifierTest, ParseBranchSystemTable) { Identifier id("db", "tbl$branch_dev$options"); ASSERT_OK_AND_ASSIGN(std::string data_table_name, id.GetDataTableName()); @@ -167,6 +182,7 @@ TEST(IdentifierTest, InvalidEmptySystemTableNameParts) { ASSERT_NOK_WITH_MSG(Identifier("db", "$options").IsSystemTable(), "Invalid table name"); ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$").IsSystemTable(), "Invalid table name"); ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_").IsSystemTable(), "Invalid table name"); + ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_ ").IsSystemTable(), "Invalid table name"); ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_dev$").IsSystemTable(), "Invalid table name"); ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$$options").IsSystemTable(), "System table can only contain one '$' separator"); diff --git a/src/paimon/core/catalog/renaming_snapshot_commit.h b/src/paimon/core/catalog/renaming_snapshot_commit.h index bda2aad41..b70ef0a80 100644 --- a/src/paimon/core/catalog/renaming_snapshot_commit.h +++ b/src/paimon/core/catalog/renaming_snapshot_commit.h @@ -24,8 +24,10 @@ #include #include +#include "fmt/format.h" #include "paimon/core/catalog/snapshot_commit.h" #include "paimon/core/snapshot.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/fs/file_system.h" #include "paimon/result.h" @@ -46,9 +48,17 @@ class RenamingSnapshotCommit : public SnapshotCommit { : fs_(fs), snapshot_manager_(snapshot_manager) {} /// @note The atomic rename detects conflicts by snapshot ID, so `base_snapshot_uuid` is unused. + /// @note `branch` has to be the branch of the snapshot manager this commit was built with. Result Commit(const std::optional& base_snapshot_uuid, - const Snapshot& snapshot, + const Snapshot& snapshot, const std::string& branch, const std::vector& statistics) override { + const std::string normalized_branch = BranchManager::NormalizeBranch(branch); + if (normalized_branch != snapshot_manager_->Branch()) { + return Status::Invalid(fmt::format( + "renaming snapshot commit built for branch '{}' cannot commit snapshot #{} to " + "branch '{}'", + snapshot_manager_->Branch(), snapshot.Id(), normalized_branch)); + } PAIMON_ASSIGN_OR_RAISE(std::string json_str, snapshot.ToJsonString()); std::string snapshot_path = snapshot_manager_->SnapshotPath(snapshot.Id()); PAIMON_ASSIGN_OR_RAISE(bool is_exist, fs_->Exists(snapshot_path)); diff --git a/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp b/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp index e2768b1c0..4f363d74c 100644 --- a/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp +++ b/src/paimon/core/catalog/renaming_snapshot_commit_test.cpp @@ -19,10 +19,12 @@ #include "paimon/core/catalog/renaming_snapshot_commit.h" #include +#include #include "gtest/gtest.h" #include "paimon/common/utils/path_util.h" #include "paimon/core/partition/partition_statistics.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/testing/utils/snapshot_test_helper.h" #include "paimon/testing/utils/testharness.h" @@ -39,7 +41,8 @@ TEST(RenamingSnapshotCommitTest, TestSimple) { ASSERT_NOK_WITH_MSG(commit->GetLastCommitTableRequest(), "renaming snapshot commit do not support get last commit table request"); Snapshot snapshot = BuildTestSnapshot(1); - ASSERT_OK_AND_ASSIGN(bool success, commit->Commit(std::nullopt, snapshot, {})); + ASSERT_OK_AND_ASSIGN(bool success, commit->Commit(std::nullopt, snapshot, + BranchManager::DEFAULT_MAIN_BRANCH, {})); ASSERT_TRUE(success); ASSERT_OK_AND_ASSIGN(bool exist, fs->Exists(PathUtil::JoinPath(dir->Str(), "snapshot/snapshot-1"))); @@ -48,8 +51,59 @@ TEST(RenamingSnapshotCommitTest, TestSimple) { fs->Exists(PathUtil::JoinPath(dir->Str(), "snapshot/LATEST"))); ASSERT_TRUE(exist1); // duplicate commit for snapshot-1 - ASSERT_OK_AND_ASSIGN(bool success1, commit->Commit("snapshot-uuid-0", snapshot, {})); + ASSERT_OK_AND_ASSIGN(bool success1, commit->Commit("snapshot-uuid-0", snapshot, + BranchManager::DEFAULT_MAIN_BRANCH, {})); ASSERT_FALSE(success1); } +TEST(RenamingSnapshotCommitTest, TestCommitToBranch) { + auto fs = std::make_shared(); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto commit = std::make_shared( + fs, std::make_shared(fs, dir->Str(), "dev")); + Snapshot snapshot = BuildTestSnapshot(1); + + ASSERT_OK_AND_ASSIGN(bool success, commit->Commit(std::nullopt, snapshot, "dev", {})); + ASSERT_TRUE(success); + const std::string branch_dir = PathUtil::JoinPath(dir->Str(), "branch/branch-dev/snapshot"); + ASSERT_OK_AND_ASSIGN(bool exist, fs->Exists(PathUtil::JoinPath(branch_dir, "snapshot-1"))); + ASSERT_TRUE(exist); + ASSERT_OK_AND_ASSIGN(bool hint_exist, fs->Exists(PathUtil::JoinPath(branch_dir, "LATEST"))); + ASSERT_TRUE(hint_exist); + ASSERT_OK_AND_ASSIGN(bool main_snapshot_dir_exists, + fs->Exists(PathUtil::JoinPath(dir->Str(), "snapshot"))); + ASSERT_FALSE(main_snapshot_dir_exists); + ASSERT_OK_AND_ASSIGN(bool duplicate, commit->Commit(std::nullopt, snapshot, "dev", {})); + ASSERT_FALSE(duplicate); +} + +TEST(RenamingSnapshotCommitTest, TestBranchHasToMatchSnapshotManager) { + auto fs = std::make_shared(); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto main_commit = std::make_shared( + fs, std::make_shared(fs, dir->Str())); + auto branch_commit = std::make_shared( + fs, std::make_shared(fs, dir->Str(), "dev")); + Snapshot snapshot = BuildTestSnapshot(1); + + ASSERT_NOK_WITH_MSG(main_commit->Commit(std::nullopt, snapshot, "dev", {}), + "renaming snapshot commit built for branch 'main' cannot commit snapshot " + "#1 to branch 'dev'"); + ASSERT_NOK_WITH_MSG( + branch_commit->Commit(std::nullopt, snapshot, BranchManager::DEFAULT_MAIN_BRANCH, {}), + "renaming snapshot commit built for branch 'dev' cannot commit snapshot #1 to branch " + "'main'"); + ASSERT_OK_AND_ASSIGN(bool branch_dir_exists, + fs->Exists(PathUtil::JoinPath(dir->Str(), "branch"))); + ASSERT_FALSE(branch_dir_exists); + ASSERT_OK_AND_ASSIGN(bool main_snapshot_dir_exists, + fs->Exists(PathUtil::JoinPath(dir->Str(), "snapshot"))); + ASSERT_FALSE(main_snapshot_dir_exists); + + ASSERT_OK_AND_ASSIGN(bool main_success, main_commit->Commit(std::nullopt, snapshot, "", {})); + ASSERT_TRUE(main_success); +} + } // namespace paimon::test diff --git a/src/paimon/core/catalog/snapshot_commit.h b/src/paimon/core/catalog/snapshot_commit.h index 08664f67b..9500205cb 100644 --- a/src/paimon/core/catalog/snapshot_commit.h +++ b/src/paimon/core/catalog/snapshot_commit.h @@ -34,9 +34,10 @@ class SnapshotCommit { /// @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. /// @param statistics Partition statistics for this change. virtual Result Commit(const std::optional& base_snapshot_uuid, - const Snapshot& snapshot, + const Snapshot& snapshot, const std::string& branch, const std::vector& statistics) = 0; virtual Result GetLastCommitTableRequest() = 0; diff --git a/src/paimon/core/operation/commit_context.cpp b/src/paimon/core/operation/commit_context.cpp index f4e80c404..e61ec950f 100644 --- a/src/paimon/core/operation/commit_context.cpp +++ b/src/paimon/core/operation/commit_context.cpp @@ -26,6 +26,7 @@ #include "fmt/format.h" #include "paimon/catalog/catalog.h" #include "paimon/common/utils/path_util.h" +#include "paimon/core/catalog/catalog_utils.h" #include "paimon/core/utils/branch_manager.h" #include "paimon/defs.h" #include "paimon/executor.h" @@ -195,22 +196,32 @@ Result> CommitContextBuilder::Finish() { if (impl_->catalog_ == nullptr && impl_->identifier_) { return Status::Invalid("cannot commit through a null catalog"); } - // Validate both branch sources: metadata paths are built for the main branch. - if (impl_->catalog_ != nullptr) { - PAIMON_ASSIGN_OR_RAISE(std::optional identifier_branch, - impl_->identifier_.value().GetBranchName()); - auto branch_option = impl_->options_.find(Options::BRANCH); - std::optional option_branch = - branch_option == impl_->options_.end() - ? std::nullopt - : std::optional(branch_option->second); - for (const std::optional& branch : {identifier_branch, option_branch}) { - if (branch && - !BranchManager::IsMainBranch(BranchManager::NormalizeBranch(branch.value()))) { + if (impl_->identifier_) { + // Before the branch is read out of the identifier: the branch of `tbl$branch_dev$options` + // reads back as `dev`, so a commit built for such a name would be aimed at another branch. + PAIMON_RETURN_NOT_OK( + CatalogUtils::CheckNotSystemTable(impl_->identifier_.value(), "commit")); + } + PAIMON_ASSIGN_OR_RAISE(std::string branch, BranchManager::ResolveBranch( + impl_->identifier_, impl_->options_, + /*explicit_branch=*/std::nullopt, "commit")); + if (!BranchManager::IsMainBranch(branch)) { + // A commit which goes to a catalog, or is handed back as a request for one, names its + // branch by an object name, so the branch has to be one that reads back as itself. + // Refused here rather than at the commit, which writes metadata before publishing it. + if (impl_->catalog_ != nullptr || impl_->use_rest_catalog_commit_) { + PAIMON_RETURN_NOT_OK(BranchManager::CheckCatalogAddressableBranch(branch)); + } + if (impl_->catalog_ != nullptr) { + PAIMON_ASSIGN_OR_RAISE(std::optional identifier_branch, + impl_->identifier_.value().GetBranchName()); + if (!identifier_branch) { + PAIMON_ASSIGN_OR_RAISE(std::string table_name, + impl_->identifier_.value().GetDataTableName()); return Status::Invalid(fmt::format( - "a commit through a catalog reads the main branch of the table, so it cannot " - "be aimed at branch '{}'", - branch.value())); + "a commit through a catalog addresses a branch by the table identifier, so " + "name branch '{}' there as '{}$branch_{}' rather than only in the '{}' option", + branch, table_name, branch, Options::BRANCH)); } } } diff --git a/src/paimon/core/operation/commit_context_test.cpp b/src/paimon/core/operation/commit_context_test.cpp index 2d3915c67..51fcab399 100644 --- a/src/paimon/core/operation/commit_context_test.cpp +++ b/src/paimon/core/operation/commit_context_test.cpp @@ -128,36 +128,95 @@ TEST(CommitContextTest, TestWithCatalog) { .Finish(), "either goes to the catalog"); + ASSERT_OK_AND_ASSIGN(auto next_ctx, builder.Finish()); + ASSERT_EQ(next_ctx->GetCatalog(), nullptr); + ASSERT_FALSE(next_ctx->GetIdentifier().has_value()); + ASSERT_EQ(next_ctx->GetTableId(), std::nullopt); +} + +TEST(CommitContextTest, TestCatalogAddressesBranchByIdentifier) { + auto catalog = std::make_shared(); + CommitContextBuilder branch_builder("table_root_path", "commit_user_1"); - ASSERT_NOK_WITH_MSG( - branch_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_dev")).Finish(), - "cannot be aimed at branch 'dev'"); - CommitContextBuilder branch_option_builder("table_root_path", "commit_user_1"); - ASSERT_NOK_WITH_MSG(branch_option_builder.WithCatalog(catalog, Identifier("db1", "t1")) - .AddOption(Options::BRANCH, "dev") - .Finish(), - "cannot be aimed at branch 'dev'"); + ASSERT_OK_AND_ASSIGN( + auto branch_ctx, + branch_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_dev")).Finish()); + ASSERT_EQ(branch_ctx->GetIdentifier().value(), Identifier("db1", "t1$branch_dev")); + CommitContextBuilder agreeing_builder("table_root_path", "commit_user_1"); + ASSERT_OK(agreeing_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_dev")) + .AddOption(Options::BRANCH, "dev") + .Finish()); + CommitContextBuilder constructed_builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN( + auto constructed_ctx, + constructed_builder.WithCatalog(catalog, Identifier("db1", "t1", "dev")).Finish()); + ASSERT_EQ(constructed_ctx->GetIdentifier().value(), Identifier("db1", "t1$branch_dev")); + CommitContextBuilder main_builder("table_root_path", "commit_user_1"); + ASSERT_OK(main_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_main")) + .AddOption(Options::BRANCH, "main") + .Finish()); + CommitContextBuilder mixed_builder("table_root_path", "commit_user_1"); ASSERT_NOK_WITH_MSG(mixed_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_dev")) .AddOption(Options::BRANCH, "main") .Finish(), - "cannot be aimed at branch 'dev'"); + "but both 'dev' and 'main' were named"); CommitContextBuilder mixed_option_builder("table_root_path", "commit_user_1"); ASSERT_NOK_WITH_MSG( mixed_option_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_main")) .AddOption(Options::BRANCH, "dev") .Finish(), - "cannot be aimed at branch 'dev'"); + "but both 'main' and 'dev' were named"); - CommitContextBuilder main_builder("table_root_path", "commit_user_1"); - ASSERT_OK(main_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_main")) - .AddOption(Options::BRANCH, "main") - .Finish()); + CommitContextBuilder option_only_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG(option_only_builder.WithCatalog(catalog, Identifier("db1", "t1")) + .AddOption(Options::BRANCH, "dev") + .Finish(), + "name branch 'dev' there as 't1$branch_dev'"); - ASSERT_OK_AND_ASSIGN(auto next_ctx, builder.Finish()); - ASSERT_EQ(next_ctx->GetCatalog(), nullptr); - ASSERT_FALSE(next_ctx->GetIdentifier().has_value()); - ASSERT_EQ(next_ctx->GetTableId(), std::nullopt); + CommitContextBuilder upper_main_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG( + upper_main_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_MAIN")).Finish(), + "a catalog names branch 'MAIN' as it names the main branch"); + + CommitContextBuilder splitter_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG(splitter_builder.WithCatalog(catalog, Identifier("db1", "t1")) + .AddOption(Options::BRANCH, "dev$options") + .Finish(), + "a branch a catalog addresses cannot contain '$'"); + + CommitContextBuilder system_table_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG( + system_table_builder.WithCatalog(catalog, Identifier("db1", "t1", "dev$options")).Finish(), + "Cannot 'commit' for system table"); +} + +TEST(CommitContextTest, TestBranch) { + CommitContextBuilder file_system_builder("table_root_path", "commit_user_1"); + ASSERT_OK(file_system_builder.AddOption(Options::BRANCH, "dev").Finish()); + + CommitContextBuilder upper_main_builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN(auto upper_main_ctx, + upper_main_builder.AddOption(Options::BRANCH, "MAIN").Finish()); + ASSERT_EQ(upper_main_ctx->GetOptions().at(Options::BRANCH), "MAIN"); + CommitContextBuilder splitter_builder("table_root_path", "commit_user_1"); + ASSERT_OK(splitter_builder.AddOption(Options::BRANCH, "dev$options").Finish()); + + CommitContextBuilder rest_builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN( + auto rest_ctx, + rest_builder.UseRESTCatalogCommit(true).AddOption(Options::BRANCH, "dev").Finish()); + ASSERT_TRUE(rest_ctx->UseRESTCatalogCommit()); + ASSERT_EQ(rest_ctx->GetOptions().at(Options::BRANCH), "dev"); + CommitContextBuilder rest_upper_main_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG(rest_upper_main_builder.UseRESTCatalogCommit(true) + .AddOption(Options::BRANCH, "MAIN") + .Finish(), + "a catalog names branch 'MAIN' as it names the main branch"); + + CommitContextBuilder escaping_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG(escaping_builder.AddOption(Options::BRANCH, "dev/../../outside").Finish(), + "branch name cannot contain path separators"); } TEST(CommitContextTest, TestSetOptionsOverridesAddedOptions) { diff --git a/src/paimon/core/operation/expire_snapshots.cpp b/src/paimon/core/operation/expire_snapshots.cpp index 9915efd46..b16f70f7c 100644 --- a/src/paimon/core/operation/expire_snapshots.cpp +++ b/src/paimon/core/operation/expire_snapshots.cpp @@ -24,9 +24,12 @@ #include #include #include +#include #include +#include #include "fmt/format.h" +#include "fmt/ranges.h" #include "paimon/common/data/binary_row.h" #include "paimon/common/executor/future.h" #include "paimon/common/utils/date_time_utils.h" @@ -42,6 +45,7 @@ #include "paimon/core/manifest/manifest_list.h" #include "paimon/core/operation/commit/realtime_commit_properties.h" #include "paimon/core/snapshot.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/fs/file_system.h" @@ -86,6 +90,25 @@ Result ExpireSnapshots::Expire() { if (snapshot_manager_ == nullptr) { return Status::Invalid("Expire failed: snapshot manager is nullptr"); } + // All branches of a table share its data files, while expiration only reads the retained + // snapshots of its own branch, so it could delete a file another branch still refers to. + if (!BranchManager::IsMainBranch(snapshot_manager_->Branch())) { + return Status::NotImplemented(fmt::format( + "Expire failed: expiring snapshots of table {} on branch '{}' is not supported, " + "because the branches of a table share its data files", + snapshot_manager_->RootPath(), snapshot_manager_->Branch())); + } + PAIMON_ASSIGN_OR_RAISE(std::vector branches, + BranchManager::ListBranches(fs_, snapshot_manager_->RootPath())); + branches.erase( + std::remove(branches.begin(), branches.end(), BranchManager::DEFAULT_MAIN_BRANCH), + branches.end()); + if (!branches.empty()) { + return Status::NotImplemented(fmt::format( + "Expire failed: expiring snapshots of table {} is not supported, because the table " + "has branches other than main ({}), which share its data files", + snapshot_manager_->RootPath(), fmt::join(branches, ", "))); + } PAIMON_ASSIGN_OR_RAISE(std::optional latest_snapshot_id, snapshot_manager_->LatestSnapshotIdFromFileSystem()); if (latest_snapshot_id == std::nullopt) { @@ -186,8 +209,8 @@ Result ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, int64 break; } } - PAIMON_LOG_DEBUG(logger_, "Snapshot expire range is [%ld, %ld]", begin_inclusive_id, - end_exclusive_id); + PAIMON_LOG_DEBUG(logger_, "Snapshot expire range of table %s is [%ld, %ld]", + snapshot_manager_->RootPath().c_str(), begin_inclusive_id, end_exclusive_id); // Since the data file deletion information for each snapshot is recorded in the delta part of // the next snapshot, it is necessary to check the next snapshot. Otherwise, its data files will diff --git a/src/paimon/core/operation/file_store_commit.cpp b/src/paimon/core/operation/file_store_commit.cpp index 15ada3fe7..2fae23baf 100644 --- a/src/paimon/core/operation/file_store_commit.cpp +++ b/src/paimon/core/operation/file_store_commit.cpp @@ -25,11 +25,13 @@ #include #include +#include "fmt/format.h" #include "paimon/catalog/catalog.h" #include "paimon/commit_context.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/binary_row_partition_computer.h" #include "paimon/core/catalog/catalog_snapshot_commit.h" +#include "paimon/core/catalog/catalog_utils.h" #include "paimon/core/catalog/renaming_snapshot_commit.h" #include "paimon/core/catalog/snapshot_commit.h" #include "paimon/core/catalog/version_managed_catalog.h" @@ -51,6 +53,7 @@ #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/snapshot_manager.h" +#include "paimon/defs.h" #include "paimon/format/file_format.h" #include "paimon/fs/file_system.h" #include "paimon/result.h" @@ -110,15 +113,16 @@ bool IsCatalogCommit(const CommitContext& ctx) { } Result> NewSnapshotCommit( - const CommitContext& ctx, const std::shared_ptr& fs, + const CommitContext& ctx, const std::optional& catalog_identifier, + const std::shared_ptr& fs, const std::shared_ptr& snapshot_manager) { if (IsCatalogCommit(ctx)) { - if (!ctx.GetIdentifier()) { + if (!catalog_identifier) { return Status::Invalid("a catalog commit requires a table identifier"); } // Reuse the caller's UUID so table recreation cannot redirect this commit. return std::shared_ptr(std::make_shared( - ctx.GetCatalog(), ctx.GetIdentifier().value(), ctx.GetTableId())); + ctx.GetCatalog(), catalog_identifier.value(), ctx.GetTableId())); } if (ctx.UseRESTCatalogCommit()) { return std::shared_ptr( @@ -196,19 +200,35 @@ Result> FileStoreCommit::Create( PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options, CoreOptions::FromMap(ctx->GetOptions(), specific_fs)); const std::string& root_path = ctx->GetRootPath(); - // Catalog-managed schemas may be absent from the table directory. + // The branch the context was built for, already validated by `CommitContextBuilder`. + PAIMON_ASSIGN_OR_RAISE(std::string branch, BranchManager::ResolveBranch( + ctx->GetIdentifier(), ctx->GetOptions(), + /*explicit_branch=*/std::nullopt, "commit")); + std::optional catalog_identifier; + if (ctx->GetIdentifier()) { + PAIMON_ASSIGN_OR_RAISE( + Identifier branch_identifier, + CatalogUtils::BranchIdentifier(ctx->GetIdentifier().value(), branch)); + catalog_identifier.emplace(std::move(branch_identifier)); + } + // Catalog-managed schemas may be absent from the table directory, but only on the main branch: + // a branch publishes its own under `branch/branch-`, which is where the load below and a + // read of the branch both read it, so a catalog answering for no branch schema does not fail + // a commit to one. std::optional catalog_table_schema; FileStoreCommitImpl::SchemaIdLoader schema_id_loader; if (ctx->GetCatalog() != nullptr) { - if (!ctx->GetIdentifier()) { + if (!catalog_identifier) { return Status::Invalid("a catalog commit requires a table identifier"); } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr loaded_schema, - ctx->GetCatalog()->LoadTableSchema(ctx->GetIdentifier().value())); - PAIMON_ASSIGN_OR_RAISE(std::string schema_json, loaded_schema->GetJsonSchema()); - catalog_table_schema = std::move(schema_json); + if (BranchManager::IsMainBranch(branch)) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr loaded_schema, + ctx->GetCatalog()->LoadTableSchema(catalog_identifier.value())); + PAIMON_ASSIGN_OR_RAISE(std::string schema_json, loaded_schema->GetJsonSchema()); + catalog_table_schema = std::move(schema_json); + } std::shared_ptr catalog = ctx->GetCatalog(); - Identifier identifier = ctx->GetIdentifier().value(); + Identifier identifier = catalog_identifier.value(); schema_id_loader = [catalog, identifier]() -> Result { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, catalog->LoadTableSchema(identifier)); @@ -222,26 +242,29 @@ Result> FileStoreCommit::Create( }; } // A format table commits by renaming files into place, so it never reaches the snapshot path - // below. The managed path here reads the main branch, so this reads the same one: the two + // below. The managed path here reads the commit's branch, so this reads the same one: the two // must not dispatch on different schemas. - auto schema_manager = std::make_shared(tmp_options.GetFileSystem(), root_path); + auto schema_manager = + std::make_shared(tmp_options.GetFileSystem(), root_path, branch); std::shared_ptr latest_schema; - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr format_table, - FormatTableLoader::TryLoad(tmp_options.GetFileSystem(), root_path, - BranchManager::DEFAULT_MAIN_BRANCH, ctx->GetOptions(), - catalog_table_schema, schema_manager.get(), &latest_schema)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr format_table, + FormatTableLoader::TryLoad( + tmp_options.GetFileSystem(), root_path, branch, ctx->GetOptions(), + catalog_table_schema, schema_manager.get(), &latest_schema)); if (format_table != nullptr) { return NewFormatTableCommit(format_table, *ctx); } if (latest_schema == nullptr) { - return Status::Invalid("not found latest schema"); + return Status::Invalid(fmt::format("not found latest schema in branch {}", branch)); } const std::shared_ptr& schema = latest_schema; auto opts = schema->Options(); for (const auto& [key, value] : ctx->GetOptions()) { opts[key] = value; } + // The metadata paths below are built for this branch, so what reads through these options, + // whose cache keys are scoped by it, has to read the same one. + opts[Options::BRANCH] = branch; std::shared_ptr arrow_schema = DataField::ConvertDataFieldsToArrowSchema(schema->Fields()); PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(opts, specific_fs)); @@ -272,11 +295,12 @@ Result> FileStoreCommit::Create( options.LegacyPartitionNameEnabled(), external_paths, global_index_external_path, options.IndexFileInDataFileDir(), ctx->GetMemoryPool())); - auto snapshot_manager = std::make_shared(options.GetFileSystem(), root_path); + auto snapshot_manager = + std::make_shared(options.GetFileSystem(), root_path, branch); if (IsCatalogCommit(*ctx)) { std::shared_ptr catalog = ctx->GetCatalog(); VersionManagedCatalog* versioned = AsVersionManaged(catalog); - Identifier identifier = ctx->GetIdentifier().value(); + Identifier identifier = catalog_identifier.value(); // Capture the catalog to keep the versioned interface alive. snapshot_manager->SetSnapshotLoader( [catalog, versioned, identifier]() -> Result> { @@ -318,8 +342,9 @@ Result> FileStoreCommit::Create( ctx->GetExecutor(), ctx->GetMemoryPool()); } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot_commit, - NewSnapshotCommit(*ctx, options.GetFileSystem(), snapshot_manager)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr snapshot_commit, + NewSnapshotCommit(*ctx, catalog_identifier, options.GetFileSystem(), snapshot_manager)); return std::make_unique( ctx->GetMemoryPool(), ctx->GetExecutor(), arrow_schema, root_path, ctx->GetCommitUser(), diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp b/src/paimon/core/operation/file_store_commit_impl.cpp index 06a1eba43..24a50e10c 100644 --- a/src/paimon/core/operation/file_store_commit_impl.cpp +++ b/src/paimon/core/operation/file_store_commit_impl.cpp @@ -77,6 +77,7 @@ #include "paimon/core/schema/table_schema.h" #include "paimon/core/table/bucket_mode.h" #include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/duration.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/partition_utils.h" @@ -568,8 +569,9 @@ Status FileStoreCommitImpl::Overwrite( std::optional watermark) { std::shared_ptr committable = CreateManifestCommittable(identifier, commit_messages, watermark, /*properties=*/{}); - PAIMON_LOG_INFO(logger_, "Ready to overwrite to table %s, number of commit messages: %zu", - table_name_.c_str(), committable->FileCommittables().size()); + PAIMON_LOG_INFO(logger_, "Ready to overwrite to table %s%s, number of commit messages: %zu", + table_name_.c_str(), CommitTargetSuffix().c_str(), + committable->FileCommittables().size()); PAIMON_ASSIGN_OR_RAISE(std::string committable_str, committable->ToString()); std::string partition_str = fmt::format("{}", partition); std::string properties_str = fmt::format("{}", committable->Properties()); @@ -593,8 +595,8 @@ Status FileStoreCommitImpl::Overwrite( report_changes.append_changelog.clear(); report_changes.compact_changelog.clear(); ScopeGuard report_guard([&]() { - PAIMON_LOG_INFO(logger_, "Finished overwrite to table %s, duration %ld ms", - table_name_.c_str(), duration.Get()); + PAIMON_LOG_INFO(logger_, "Finished overwrite to table %s%s, duration %ld ms", + table_name_.c_str(), CommitTargetSuffix().c_str(), duration.Get()); ReportCommit(report_changes, duration.Get(), generated_snapshot, attempt); }); @@ -609,8 +611,9 @@ Result FileStoreCommitImpl::FilterAndOverwrite( std::optional watermark) { std::shared_ptr committable = CreateManifestCommittable(identifier, commit_messages, watermark, /*properties=*/{}); - PAIMON_LOG_INFO(logger_, "Ready to overwrite to table %s, number of commit messages: %zu", - table_name_.c_str(), committable->FileCommittables().size()); + PAIMON_LOG_INFO(logger_, "Ready to overwrite to table %s%s, number of commit messages: %zu", + table_name_.c_str(), CommitTargetSuffix().c_str(), + committable->FileCommittables().size()); PAIMON_ASSIGN_OR_RAISE(std::string committable_str, committable->ToString()); std::string partition_str = fmt::format("{}", partition); std::string properties_str = fmt::format("{}", committable->Properties()); @@ -639,8 +642,8 @@ Result FileStoreCommitImpl::FilterAndOverwrite( report_changes.append_changelog.clear(); report_changes.compact_changelog.clear(); ScopeGuard report_guard([&]() { - PAIMON_LOG_INFO(logger_, "Finished overwrite to table %s, duration %ld ms", - table_name_.c_str(), duration.Get()); + PAIMON_LOG_INFO(logger_, "Finished overwrite to table %s%s, duration %ld ms", + table_name_.c_str(), CommitTargetSuffix().c_str(), duration.Get()); ReportCommit(report_changes, duration.Get(), generated_snapshot, attempt); }); @@ -650,8 +653,8 @@ Result FileStoreCommitImpl::FilterAndOverwrite( } else { // Align with Java: filtered duplicate is treated as one resolved commit attempt. attempt = 1; - PAIMON_LOG_INFO(logger_, "Finished overwrite to table %s, duration %ld ms", - table_name_.c_str(), duration.Get()); + PAIMON_LOG_INFO(logger_, "Finished overwrite to table %s%s, duration %ld ms", + table_name_.c_str(), CommitTargetSuffix().c_str(), duration.Get()); } return actual_committables.size(); } @@ -858,8 +861,9 @@ Result FileStoreCommitImpl::TryOverwrite( Status FileStoreCommitImpl::Commit( const std::shared_ptr& committable, bool check_append_files, bool retry_on_conflict, const std::map& realtime_ranges) { - PAIMON_LOG_INFO(logger_, "Ready to commit to table %s, number of commit messages: %zu", - table_name_.c_str(), committable->FileCommittables().size()); + PAIMON_LOG_INFO(logger_, "Ready to commit to table %s%s, number of commit messages: %zu", + table_name_.c_str(), CommitTargetSuffix().c_str(), + committable->FileCommittables().size()); PAIMON_ASSIGN_OR_RAISE(std::string committable_str, committable->ToString()); PAIMON_LOG_DEBUG(logger_, "Ready to commit\n%s", committable_str.c_str()); @@ -871,8 +875,8 @@ Status FileStoreCommitImpl::Commit( CollectChanges(committable->FileCommittables())); ScopeGuard report_guard([&]() { PAIMON_LOG_INFO(logger_, - "Finished (Uncertain of success) commit to table %s, duration %ld ms", - table_name_.c_str(), duration.Get()); + "Finished (Uncertain of success) commit to table %s%s, duration %ld ms", + table_name_.c_str(), CommitTargetSuffix().c_str(), duration.Get()); ReportCommit(changes, duration.Get(), generated_snapshot, attempt); }); @@ -1346,9 +1350,12 @@ Result FileStoreCommitImpl::TryCommitOnce( latest_snapshot ? latest_snapshot.value().Statistics() : std::nullopt; std::optional changelog_record_count = ManifestEntry::NullableRecordCount(changelog_entries); - // Reload on each attempt to account for concurrent schema changes. + // Reload on each attempt to account for concurrent schema changes. A branch publishes the + // schema this commit writes with under the branch, which is also where a read of the branch + // resolves the id recorded here, so it is read there rather than from the catalog; only the + // main branch may have a current schema the catalog alone holds. int64_t schema_id = table_schema_->Id(); - if (schema_id_loader_) { + if (schema_id_loader_ && BranchManager::IsMainBranch(snapshot_manager_->Branch())) { PAIMON_ASSIGN_OR_RAISE(schema_id, schema_id_loader_()); } else { PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema, @@ -1448,7 +1455,17 @@ Status FileStoreCommitImpl::CheckRowIdSchemasArePublished( } std::string FileStoreCommitImpl::CommitTargetSuffix() const { - const std::string target = snapshot_commit_->DescribeTarget(); + // The snapshot manager's branch is the one handed to `SnapshotCommit::Commit()`, so a log + // line names the branch the snapshot went to rather than the one the commit was built around. + std::string target; + const std::string& branch = snapshot_manager_->Branch(); + if (!BranchManager::IsMainBranch(branch)) { + target = "branch " + branch; + } + const std::string described = snapshot_commit_->DescribeTarget(); + if (!described.empty()) { + target = target.empty() ? described : target + ", " + described; + } return target.empty() ? std::string() : " (" + target + ")"; } @@ -1471,8 +1488,8 @@ Result FileStoreCommitImpl::CommitSnapshotImpl( } std::optional base_snapshot_uuid = base_snapshot ? base_snapshot.value().Uuid() : std::nullopt; - Result commit_result = - snapshot_commit_->Commit(base_snapshot_uuid, new_snapshot, statistics); + Result commit_result = snapshot_commit_->Commit(base_snapshot_uuid, new_snapshot, + snapshot_manager_->Branch(), statistics); if (!commit_result.ok()) { // An error leaves publication uncertain; preserve metadata and report the attempted UUID. return Status::Invalid(fmt::format( 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 b6710ba81..07207dc38 100644 --- a/src/paimon/core/operation/file_store_commit_impl_test.cpp +++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp @@ -68,10 +68,12 @@ #include "paimon/core/operation/commit/realtime_commit_properties.h" #include "paimon/core/operation/metrics/commit_metrics.h" #include "paimon/core/partition/partition_statistics.h" +#include "paimon/core/schema/schema_manager.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/stats/simple_stats.h" #include "paimon/core/table/sink/commit_message_impl.h" #include "paimon/core/table/system/system_table_schema.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/file_utils.h" #include "paimon/core/utils/snapshot_manager.h" @@ -282,13 +284,16 @@ class FileStoreCommitImplTest : public testing::Test { return catalog; } + static std::vector FirstCommitMessageFiles() { + return {"/f1=10/bucket-0/data-51a45441-6037-4af3-b67b-5cefd75dc6f2-0.orc", + "/f1=10/bucket-1/data-6828284c-e707-49b5-af6b-69be79af120c-0.orc", + "/f1=20/bucket-0/data-8dc7f04c-3c98-48b2-9d56-834d746c4a40-0.orc", + "/f1=10/bucket-1/data-fd1d2255-43f2-4534-b4cc-08b29e662940-0.orc", + "/f1=20/bucket-0/data-7b3f4cc7-116b-4d2f-9c62-5dadc1f11bcb-0.orc"}; + } + Status PrepareFirstCommitMessageFiles() { - return PrepareFakeFiles( - {"/f1=10/bucket-0/data-51a45441-6037-4af3-b67b-5cefd75dc6f2-0.orc", - "/f1=10/bucket-1/data-6828284c-e707-49b5-af6b-69be79af120c-0.orc", - "/f1=20/bucket-0/data-8dc7f04c-3c98-48b2-9d56-834d746c4a40-0.orc", - "/f1=10/bucket-1/data-fd1d2255-43f2-4534-b4cc-08b29e662940-0.orc", - "/f1=20/bucket-0/data-7b3f4cc7-116b-4d2f-9c62-5dadc1f11bcb-0.orc"}); + return PrepareFakeFiles(FirstCommitMessageFiles()); } void PublishAcceptedSnapshots(const std::shared_ptr& catalog) { @@ -329,7 +334,7 @@ class FileStoreCommitImplTest : public testing::Test { std::shared_ptr catalog; std::string commit_user = "commit_user_1"; std::string root_path; - Identifier identifier{"db", "tbl"}; + std::optional identifier = Identifier("db", "tbl"); std::optional table_id; bool append_commit_check_conflict = false; std::map options; @@ -340,7 +345,7 @@ class FileStoreCommitImplTest : public testing::Test { CommitContextBuilder builder(spec.root_path.empty() ? table_path_ : spec.root_path, spec.commit_user); builder.AddOption(Options::FILE_SYSTEM, "local") - .WithCatalog(spec.catalog, spec.identifier) + .WithCatalog(spec.catalog, spec.identifier.value()) .AppendCommitCheckConflict(spec.append_commit_check_conflict); if (spec.table_id) { builder.WithTableId(spec.table_id.value()); @@ -404,6 +409,51 @@ class FileStoreCommitImplTest : public testing::Test { ASSERT_EQ(recovery->catalog->AcceptedSnapshots().back().CommitUser(), "commit_user_2"); } + void CheckExpireIsRefusedWithoutDeleting(const std::string& branch, + const std::string& expected_msg) { + CommitContextBuilder builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, + builder.AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::BRANCH, branch) + .AddOption(Options::SNAPSHOT_NUM_RETAINED_MIN, "1") + .AddOption(Options::SNAPSHOT_NUM_RETAINED_MAX, "1") + .AddOption(Options::SNAPSHOT_TIME_RETAINED, "1ms") + .Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + FileStoreCommit::Create(std::move(context))); + ASSERT_OK(PrepareFirstCommitMessageFiles()); + ASSERT_OK(commit->Commit(CommitMessagesOfRound(1), /*commit_identifier=*/1)); + ASSERT_OK(commit->DropPartition({{{"f1", "10"}}}, /*commit_identifier=*/2)); + + SnapshotManager snapshot_manager(file_system_, table_path_, branch); + ASSERT_OK_AND_ASSIGN(Snapshot first, snapshot_manager.LoadSnapshot(1)); + const std::string manifest_dir = PathUtil::JoinPath(table_path_, "manifest"); + size_t manifest_count = CountFiles(manifest_dir); + ASSERT_GT(manifest_count, 0u); + + Result expired = commit->Expire(); + ASSERT_TRUE(expired.status().IsNotImplemented()) << expired.status().ToString(); + ASSERT_NOK_WITH_MSG(expired, expected_msg); + + ASSERT_OK_AND_ASSIGN(bool snapshot_exists, snapshot_manager.SnapshotExists(1)); + ASSERT_TRUE(snapshot_exists); + ASSERT_OK_AND_ASSIGN(std::optional earliest_id, + snapshot_manager.EarliestSnapshotId()); + ASSERT_EQ(earliest_id, std::optional(1)); + for (const std::string& manifest_list : + {first.BaseManifestList(), first.DeltaManifestList()}) { + ASSERT_OK_AND_ASSIGN(bool manifest_list_exists, file_system_->Exists(PathUtil::JoinPath( + manifest_dir, manifest_list))); + ASSERT_TRUE(manifest_list_exists) << manifest_list; + } + ASSERT_EQ(CountFiles(manifest_dir), manifest_count); + for (const std::string& data_file : FirstCommitMessageFiles()) { + ASSERT_OK_AND_ASSIGN(bool data_file_exists, + file_system_->Exists(PathUtil::JoinPath(table_path_, data_file))); + ASSERT_TRUE(data_file_exists) << data_file; + } + } + size_t CountFiles(const std::string& dir) const { size_t count = 0; std::error_code ec; @@ -721,6 +771,193 @@ 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 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 context, + builder.AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::BRANCH, "dev") + .Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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_snapshot_exists, + file_system_->Exists(PathUtil::JoinPath(branch_snapshots, "snapshot-1"))); + ASSERT_TRUE(branch_snapshot_exists); + 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_snapshot_exists, file_system_->Exists(PathUtil::JoinPath( + table_path_, "snapshot/snapshot-1"))); + ASSERT_FALSE(main_snapshot_exists); + ASSERT_GT(CountFiles(PathUtil::JoinPath(table_path_, "manifest")), 0u); + + SnapshotManager branch_snapshot_manager(file_system_, table_path_, "dev"); + ASSERT_OK_AND_ASSIGN(std::optional 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 context, + builder.AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::BRANCH, "dev") + .UseRESTCatalogCommit(true) + .WithTableId("table-uuid") + .Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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("table-uuid")); + ASSERT_EQ(request.GetSnapshot().Id(), 1); + ASSERT_EQ(request.GetSnapshot().SchemaId(), 0); + + ASSERT_OK_AND_ASSIGN(bool branch_snapshot_exists, + file_system_->Exists(PathUtil::JoinPath( + table_path_, "branch/branch-dev/snapshot/snapshot-1"))); + ASSERT_FALSE(branch_snapshot_exists); + ASSERT_OK_AND_ASSIGN(bool main_snapshot_exists, file_system_->Exists(PathUtil::JoinPath( + table_path_, "snapshot/snapshot-1"))); + ASSERT_FALSE(main_snapshot_exists); + 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 base_context, + base_builder.AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::BRANCH, "dev") + .Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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 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 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_snapshot_exists, + file_system_->Exists(PathUtil::JoinPath( + table_path_, "branch/branch-dev/snapshot/snapshot-2"))); + ASSERT_FALSE(branch_snapshot_exists); +} + +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 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 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 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 commit, CreateCatalogCommit(spec)); + + Status failed = commit->Commit(CommitMessagesOfRound(1)); + ASSERT_NOK_WITH_MSG(failed, "catalog unreachable"); + ASSERT_NOK_WITH_MSG(failed, "(branch dev, through catalog"); +} + +TEST_F(FileStoreCommitImplTest, TestCatalogCommitCanonicalizesTheMainBranchIdentifier) { + auto catalog = CreateCatalog(); + CatalogCommitSpec spec; + spec.catalog = catalog; + spec.identifier.emplace("db", "tbl$branch_main"); + spec.table_id = "table-uuid-1"; + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, CreateCatalogCommit(spec)); + ASSERT_OK(commit->Commit(CommitMessagesOfRound(1), /*commit_identifier=*/1)); + + ASSERT_GT(catalog->LoadTableSchemaCalls(), 0u); + for (const Identifier& asked : catalog->LoadTableSchemaIdentifiers()) { + ASSERT_EQ(asked, Identifier("db", "tbl")); + } + ASSERT_GT(catalog->LoadSnapshotCalls(), 0u); + for (const Identifier& asked : catalog->LoadSnapshotIdentifiers()) { + ASSERT_EQ(asked, Identifier("db", "tbl")); + } + ASSERT_EQ(catalog->CommitCalls().size(), 1u); + ASSERT_EQ(catalog->CommitCalls().front().identifier, Identifier("db", "tbl")); +} + TEST_F(FileStoreCommitImplTest, TestCommitIsRefusedForATableRecreatedUnderTheSameName) { auto catalog = CreateCatalog(true, "table-uuid-of-the-new-table"); catalog->CheckTableUuid(); @@ -2788,6 +3025,31 @@ TEST_F(FileStoreCommitImplTest, TestDropPartitionAndExpireSnapshot) { ASSERT_EQ(2, manifests[0].NumDeletedFiles()); } +TEST_F(FileStoreCommitImplTest, TestExpireOnBranchIsNotSupported) { + SchemaManager branch_schema_manager(file_system_, table_path_, "dev"); + ASSERT_OK(branch_schema_manager.CreateTable(arrow::schema(fields_), /*partition_keys=*/{"f1"}, + /*primary_keys=*/{}, /*options=*/{})); + + ASSERT_NO_FATAL_FAILURE(CheckExpireIsRefusedWithoutDeleting( + "dev", + "on branch 'dev' is not supported, because the branches of a table share its data " + "files")); +} + +TEST_F(FileStoreCommitImplTest, TestExpireOnMainOfTableWithBranchesIsNotSupported) { + for (const char* branch : {"dev", "test"}) { + SchemaManager branch_schema_manager(file_system_, table_path_, branch); + ASSERT_OK(branch_schema_manager.CreateTable(arrow::schema(fields_), + /*partition_keys=*/{"f1"}, + /*primary_keys=*/{}, /*options=*/{})); + } + + ASSERT_NO_FATAL_FAILURE(CheckExpireIsRefusedWithoutDeleting( + BranchManager::DEFAULT_MAIN_BRANCH, + "is not supported, because the table has branches other than main (dev, test), which " + "share its data files")); +} + TEST_F(FileStoreCommitImplTest, TestDropMultiPartitionAndExpireSnapshot) { CommitContextBuilder context_builder(table_path_, "commit_user_1"); ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, diff --git a/src/paimon/core/operation/file_store_write.cpp b/src/paimon/core/operation/file_store_write.cpp index 38686cbc0..02121e6b1 100644 --- a/src/paimon/core/operation/file_store_write.cpp +++ b/src/paimon/core/operation/file_store_write.cpp @@ -27,6 +27,7 @@ #include "paimon/common/data/blob_utils.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/fields_comparator.h" +#include "paimon/core/catalog/catalog_utils.h" #include "paimon/core/catalog/version_managed_catalog.h" #include "paimon/core/core_options.h" #include "paimon/core/disk/io_manager.h" @@ -46,10 +47,12 @@ #include "paimon/core/table/bucket_mode.h" #include "paimon/core/table/format/format_table_file_store_write.h" #include "paimon/core/table/format/format_table_loader.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/primary_key_table_utils.h" #include "paimon/core/utils/snapshot_manager.h" +#include "paimon/defs.h" #include "paimon/format/file_format.h" #include "paimon/result.h" #include "paimon/schema/schema.h" @@ -159,16 +162,28 @@ Result> FileStoreWrite::Create(std::unique_ptrGetOptions(), specific_fs, ctx->GetFileSystemSchemeToIdentifierMap())); + std::string branch = ctx->GetBranch(); + std::optional catalog_identifier; + if (ctx->GetIdentifier()) { + PAIMON_ASSIGN_OR_RAISE( + Identifier branch_identifier, + CatalogUtils::BranchIdentifier(ctx->GetIdentifier().value(), branch)); + catalog_identifier.emplace(std::move(branch_identifier)); + } + // Only the main branch may have a current schema the catalog alone holds. A branch publishes + // its own under `branch/branch-`, which is where the load below and a read of the branch + // both read it, so a catalog answering for no branch schema does not fail a write to one. std::optional catalog_table_schema; if (ctx->GetCatalog() != nullptr) { - if (!ctx->GetIdentifier()) { + if (!catalog_identifier) { return Status::Invalid("a catalog write requires a table identifier"); } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, - ctx->GetCatalog()->LoadTableSchema(ctx->GetIdentifier().value())); - PAIMON_ASSIGN_OR_RAISE(catalog_table_schema, schema->GetJsonSchema()); + if (BranchManager::IsMainBranch(branch)) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, + ctx->GetCatalog()->LoadTableSchema(catalog_identifier.value())); + PAIMON_ASSIGN_OR_RAISE(catalog_table_schema, schema->GetJsonSchema()); + } } - std::string branch = ctx->GetBranch(); // A format table writes plain data files into a directory, so it never reaches the manifest // path below. auto schema_manager = @@ -185,7 +200,7 @@ Result> FileStoreWrite::Create(std::unique_ptr& schema = latest_schema; auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(schema->Fields()); @@ -194,6 +209,9 @@ Result> FileStoreWrite::Create(std::unique_ptrGetOptions()) { opts[key] = value; } + // The metadata paths above are built for this branch, so what reads through these options, + // whose cache keys are scoped by it, has to read the same one. + opts[Options::BRANCH] = branch; PAIMON_ASSIGN_OR_RAISE( CoreOptions options, CoreOptions::FromMap(opts, specific_fs, ctx->GetFileSystemSchemeToIdentifierMap())); @@ -216,7 +234,7 @@ Result> FileStoreWrite::Create(std::unique_ptr(options.GetFileSystem(), ctx->GetRootPath(), branch); if (VersionManagedCatalog* versioned = AsVersionManaged(ctx->GetCatalog())) { std::shared_ptr catalog = ctx->GetCatalog(); - Identifier identifier = ctx->GetIdentifier().value(); + Identifier identifier = catalog_identifier.value(); // Keep the catalog alive while the writer uses its snapshot loader. snapshot_manager->SetSnapshotLoader( [catalog, versioned, identifier]() -> Result> { diff --git a/src/paimon/core/operation/file_store_write_test.cpp b/src/paimon/core/operation/file_store_write_test.cpp index 20d247fbd..31356a0cd 100644 --- a/src/paimon/core/operation/file_store_write_test.cpp +++ b/src/paimon/core/operation/file_store_write_test.cpp @@ -186,6 +186,42 @@ TEST(FileStoreWriteTest, TestCreateAppendTable) { } } +TEST(FileStoreWriteTest, TestCatalogWriteAsksTheCatalogOnlyForTheMainBranchSchema) { + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + const auto logical_schema = arrow::schema( + {arrow::field("id", arrow::int64(), false), arrow::field("value", arrow::utf8())}); + ASSERT_OK_AND_ASSIGN(std::shared_ptr schema, + TableSchema::Create(0, logical_schema, /*partition_keys=*/{}, + /*primary_keys=*/{}, /*options=*/{})); + auto catalog = std::make_shared(); + catalog->SetTableSchema(schema); + catalog->SetFileSystem(dir->GetFileSystem()); + + WriteContextBuilder main_builder(dir->Str(), "commit_user_1"); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr main_context, + main_builder.WithCatalog(catalog, Identifier("db", "table$branch_main")).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr main_write, + FileStoreWrite::Create(std::move(main_context))); + ASSERT_OK(main_write->Close()); + ASSERT_EQ(catalog->LoadTableSchemaIdentifiers().size(), 1u); + ASSERT_EQ(catalog->LoadTableSchemaIdentifiers().front(), Identifier("db", "table")); + + SchemaManager branch_schema_manager(dir->GetFileSystem(), dir->Str(), "dev"); + ASSERT_OK(branch_schema_manager.CreateTable(logical_schema, /*partition_keys=*/{}, + /*primary_keys=*/{}, /*options=*/{})); + catalog->SetTableSchema(nullptr); + WriteContextBuilder branch_builder(dir->Str(), "commit_user_1"); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr branch_context, + branch_builder.WithCatalog(catalog, Identifier("db", "table$branch_dev")).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr branch_write, + FileStoreWrite::Create(std::move(branch_context))); + ASSERT_OK(branch_write->Close()); + ASSERT_EQ(catalog->LoadTableSchemaIdentifiers().size(), 1u); +} + TEST(FileStoreWriteTest, TestCreateWriterForLoadedMapBlobTable) { auto dir = UniqueTestDirectory::Create(); std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp index 7545c543c..5626e8c2d 100644 --- a/src/paimon/core/operation/read_context.cpp +++ b/src/paimon/core/operation/read_context.cpp @@ -93,7 +93,7 @@ class ReadContextBuilder::Impl { public: friend class ReadContextBuilder; void Reset() { - branch_ = BranchManager::DEFAULT_MAIN_BRANCH; + branch_.reset(); read_field_names_.clear(); read_field_ids_.clear(); read_schema_.reset(); @@ -124,7 +124,9 @@ class ReadContextBuilder::Impl { /// than a setting of one read of it. std::shared_ptr format_table_; bool built_from_format_table_ = false; - std::string branch_ = BranchManager::DEFAULT_MAIN_BRANCH; + /// Unset until `WithBranch()` names one, so that a branch named only by the `branch` + /// option is not overruled by the default. + std::optional branch_; std::vector read_field_names_; std::vector read_field_ids_; std::unique_ptr read_schema_; @@ -315,7 +317,8 @@ Result> ReadContextBuilder::Finish() { "a format table carries the file system it was loaded through, so WithFileSystem() " "and WithFileSystemSchemeToIdentifierMap() cannot be used with one"); } - if (impl_->branch_ != BranchManager::DEFAULT_MAIN_BRANCH) { + if (impl_->branch_ && + !BranchManager::IsMainBranch(BranchManager::NormalizeBranch(impl_->branch_.value()))) { return Status::Invalid( "a format table has no branches, so WithBranch() cannot be used with one"); } @@ -324,8 +327,9 @@ Result> ReadContextBuilder::Finish() { if (impl_->path_.empty()) { return Status::Invalid("cannot read with empty table path"); } - // The branch names a directory under the table path, so it must stay a single path component. - PAIMON_RETURN_NOT_OK(BranchManager::CheckValidBranch(impl_->branch_)); + PAIMON_ASSIGN_OR_RAISE(std::string branch, + BranchManager::ResolveBranch(/*identifier=*/std::nullopt, + impl_->options_, impl_->branch_, "read")); if (impl_->enable_prefetch_ && impl_->prefetch_max_parallel_num_ == 0) { return Status::Invalid("prefetch max parallel num should be greater than 0"); } @@ -351,14 +355,14 @@ Result> ReadContextBuilder::Finish() { return Status::Invalid("row to batch thread number should be greater than 0"); } auto ctx = std::make_unique( - impl_->path_, impl_->branch_, impl_->read_field_names_, impl_->read_field_ids_, - impl_->predicate_, impl_->enable_predicate_filter_, impl_->enable_prefetch_, - impl_->enable_late_materializing_, impl_->prefetch_batch_count_, - impl_->prefetch_max_parallel_num_, impl_->enable_multi_thread_row_to_batch_, - impl_->row_to_batch_thread_number_, impl_->table_schema_, impl_->memory_pool_, - impl_->executor_, impl_->specific_file_system_, impl_->fs_scheme_to_identifier_map_, - impl_->realtime_context_, impl_->options_, impl_->read_ahead_cache_enabled_, - impl_->cache_config_, impl_->cache_, impl_->format_table_, impl_->warmup_level_); + impl_->path_, branch, impl_->read_field_names_, impl_->read_field_ids_, impl_->predicate_, + impl_->enable_predicate_filter_, impl_->enable_prefetch_, impl_->enable_late_materializing_, + impl_->prefetch_batch_count_, impl_->prefetch_max_parallel_num_, + impl_->enable_multi_thread_row_to_batch_, impl_->row_to_batch_thread_number_, + impl_->table_schema_, impl_->memory_pool_, impl_->executor_, impl_->specific_file_system_, + impl_->fs_scheme_to_identifier_map_, impl_->realtime_context_, impl_->options_, + impl_->read_ahead_cache_enabled_, impl_->cache_config_, impl_->cache_, impl_->format_table_, + impl_->warmup_level_); if (impl_->read_schema_ && impl_->read_schema_->release) { ctx->SetReadSchema(std::move(impl_->read_schema_)); } diff --git a/src/paimon/core/operation/read_context_test.cpp b/src/paimon/core/operation/read_context_test.cpp index 572fd5a5a..e1cdc3eb5 100644 --- a/src/paimon/core/operation/read_context_test.cpp +++ b/src/paimon/core/operation/read_context_test.cpp @@ -24,6 +24,7 @@ #include "arrow/type.h" #include "gtest/gtest.h" #include "paimon/common/io/cache/lru_cache.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/defs.h" #include "paimon/executor.h" #include "paimon/memory/memory_pool.h" @@ -149,18 +150,35 @@ TEST(ReadContextTest, TestSetOptionsOverridesAddedOptions) { ASSERT_EQ(expected_options, ctx->GetOptions()); } -TEST(ReadContextTest, TestRejectBranchLeavingTablePath) { - // The branch names a directory under the table path, so a value that is not a single path - // component is rejected when the context is built. - ReadContextBuilder builder("table_root_path"); - builder.WithBranch("rt/../../../../../outside"); - ASSERT_NOK_WITH_MSG(builder.Finish(), "branch name cannot contain path separators"); +TEST(ReadContextTest, TestBranch) { + ReadContextBuilder option_builder("table_root_path"); + ASSERT_OK_AND_ASSIGN(auto option_ctx, option_builder.AddOption(Options::BRANCH, "rt").Finish()); + ASSERT_EQ("rt", option_ctx->GetBranch()); + + ReadContextBuilder agreeing_builder("table_root_path"); + ASSERT_OK_AND_ASSIGN( + auto agreeing_ctx, + agreeing_builder.WithBranch("rt").AddOption(Options::BRANCH, "rt").Finish()); + ASSERT_EQ("rt", agreeing_ctx->GetBranch()); + + ReadContextBuilder mixed_builder("table_root_path"); + ASSERT_NOK_WITH_MSG(mixed_builder.WithBranch("rt").AddOption(Options::BRANCH, "dev").Finish(), + "but both 'dev' and 'rt' were named"); // An empty branch selects the main branch and stays accepted. ReadContextBuilder main_builder("table_root_path"); main_builder.WithBranch(""); ASSERT_OK_AND_ASSIGN(auto ctx, main_builder.Finish()); - ASSERT_EQ("", ctx->GetBranch()); + ASSERT_EQ(BranchManager::DEFAULT_MAIN_BRANCH, ctx->GetBranch()); + ReadContextBuilder empty_builder("table_root_path"); + ASSERT_NOK_WITH_MSG(empty_builder.WithBranch("").AddOption(Options::BRANCH, "dev").Finish(), + "but both 'dev' and 'main' were named"); + + // The branch names a directory under the table path, so a value that is not a single path + // component is rejected when the context is built. + ReadContextBuilder escaping_builder("table_root_path"); + escaping_builder.WithBranch("rt/../../../../../outside"); + ASSERT_NOK_WITH_MSG(escaping_builder.Finish(), "branch name cannot contain path separators"); } TEST(ReadContextTest, TestFileSystemAndSchemeMapConflict) { diff --git a/src/paimon/core/operation/write_context.cpp b/src/paimon/core/operation/write_context.cpp index a62080c01..e31d64ac4 100644 --- a/src/paimon/core/operation/write_context.cpp +++ b/src/paimon/core/operation/write_context.cpp @@ -24,8 +24,8 @@ #include "fmt/format.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/path_util.h" +#include "paimon/core/catalog/catalog_utils.h" #include "paimon/core/utils/branch_manager.h" -#include "paimon/defs.h" #include "paimon/executor.h" #include "paimon/memory/memory_pool.h" #include "paimon/result.h" @@ -81,7 +81,7 @@ class WriteContextBuilder::Impl { memory_pool_ = GetDefaultPool(); executor_ = CreateDefaultExecutor(); temp_directory_.clear(); - branch_ = BranchManager::DEFAULT_MAIN_BRANCH; + branch_.reset(); write_schema_.clear(); fs_scheme_to_identifier_map_.clear(); specific_file_system_.reset(); @@ -98,7 +98,9 @@ class WriteContextBuilder::Impl { std::shared_ptr format_table_; bool built_from_format_table_ = false; std::string commit_user_; - std::string branch_ = BranchManager::DEFAULT_MAIN_BRANCH; + /// Unset until `WithBranch()` names one, so that a branch named only by the identifier or + /// the `branch` option is not overruled by the default. + std::optional branch_; std::optional write_id_; bool is_streaming_mode_ = false; bool ignore_num_bucket_check_ = false; @@ -239,7 +241,8 @@ Result> WriteContextBuilder::Finish() { "a format table carries the file system it was loaded through, so WithFileSystem() " "and WithFileSystemSchemeToIdentifierMap() cannot be used with one"); } - if (impl_->branch_ != BranchManager::DEFAULT_MAIN_BRANCH) { + if (impl_->branch_ && + !BranchManager::IsMainBranch(BranchManager::NormalizeBranch(impl_->branch_.value()))) { return Status::Invalid( "a format table has no branches, so WithBranch() cannot be used with one"); } @@ -248,27 +251,29 @@ Result> WriteContextBuilder::Finish() { if (impl_->root_path_.empty()) { return Status::Invalid("root path is empty"); } - // The branch names a directory under the root path, so it must stay a single path component. - PAIMON_RETURN_NOT_OK(BranchManager::CheckValidBranch(impl_->branch_)); if (impl_->catalog_ == nullptr && impl_->identifier_) { return Status::Invalid("cannot write through a null catalog"); } - if (impl_->catalog_ != nullptr) { + if (impl_->identifier_) { + // Before the branch is read out of the identifier: the branch of `tbl$branch_dev$options` + // reads back as `dev`, so a write built for such a name would go to another branch. + PAIMON_RETURN_NOT_OK( + CatalogUtils::CheckNotSystemTable(impl_->identifier_.value(), "write")); + } + PAIMON_ASSIGN_OR_RAISE( + std::string branch, + BranchManager::ResolveBranch(impl_->identifier_, impl_->options_, impl_->branch_, "write")); + if (impl_->catalog_ != nullptr && !BranchManager::IsMainBranch(branch)) { + // Refused here rather than at the commit of what is written to the branch. + PAIMON_RETURN_NOT_OK(BranchManager::CheckCatalogAddressableBranch(branch)); PAIMON_ASSIGN_OR_RAISE(std::optional identifier_branch, impl_->identifier_->GetBranchName()); - auto branch_option = impl_->options_.find(Options::BRANCH); - std::optional option_branch = - branch_option == impl_->options_.end() - ? std::nullopt - : std::optional(branch_option->second); - for (const std::optional& branch : - {identifier_branch, option_branch, std::optional(impl_->branch_)}) { - if (branch && - !BranchManager::IsMainBranch(BranchManager::NormalizeBranch(branch.value()))) { - return Status::Invalid( - fmt::format("a write through a catalog requires the main branch, but got '{}'", - branch.value())); - } + if (!identifier_branch) { + PAIMON_ASSIGN_OR_RAISE(std::string table_name, impl_->identifier_->GetDataTableName()); + return Status::Invalid(fmt::format( + "a write through a catalog addresses a branch by the table identifier, so name " + "branch '{}' there as '{}$branch_{}'", + branch, table_name, branch)); } } bool enable_multi_thread_spill = impl_->spill_thread_number_ > 0; @@ -279,10 +284,10 @@ Result> WriteContextBuilder::Finish() { auto ctx = std::make_unique( impl_->root_path_, impl_->commit_user_, impl_->is_streaming_mode_, impl_->ignore_num_bucket_check_, impl_->ignore_previous_files_, enable_multi_thread_spill, - impl_->write_id_, impl_->branch_, impl_->write_schema_, impl_->memory_pool_, - impl_->executor_, impl_->temp_directory_, impl_->specific_file_system_, - impl_->fs_scheme_to_identifier_map_, impl_->realtime_context_, impl_->options_, - impl_->format_table_, impl_->catalog_, impl_->identifier_); + impl_->write_id_, branch, impl_->write_schema_, impl_->memory_pool_, impl_->executor_, + impl_->temp_directory_, impl_->specific_file_system_, impl_->fs_scheme_to_identifier_map_, + impl_->realtime_context_, impl_->options_, impl_->format_table_, impl_->catalog_, + impl_->identifier_); impl_->Reset(); return ctx; } diff --git a/src/paimon/core/operation/write_context_test.cpp b/src/paimon/core/operation/write_context_test.cpp index 1193e3364..5e777c351 100644 --- a/src/paimon/core/operation/write_context_test.cpp +++ b/src/paimon/core/operation/write_context_test.cpp @@ -21,6 +21,7 @@ #include #include "gtest/gtest.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/defs.h" #include "paimon/executor.h" #include "paimon/memory/memory_pool.h" @@ -68,21 +69,85 @@ TEST(WriteContextTest, TestWithCatalog) { "cannot write through a null catalog"); } -TEST(WriteContextTest, TestCatalogRequiresMainBranch) { +TEST(WriteContextTest, TestCatalogAddressesBranchByIdentifier) { auto catalog = std::make_shared(); - for (int32_t branch_source = 0; branch_source < 3; ++branch_source) { + WriteContextBuilder builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN(auto ctx, builder.WithCatalog(catalog, Identifier("db1", "t1$branch_dev")) + .WithBranch("dev") + .AddOption(Options::BRANCH, "dev") + .Finish()); + ASSERT_EQ(ctx->GetBranch(), "dev"); + + WriteContextBuilder identifier_builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN( + auto identifier_ctx, + identifier_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_dev")).Finish()); + ASSERT_EQ(identifier_ctx->GetBranch(), "dev"); + + for (int32_t branch_source = 0; branch_source < 2; ++branch_source) { SCOPED_TRACE(branch_source); - WriteContextBuilder builder("table_root_path", "commit_user_1"); - builder.WithCatalog(catalog, Identifier("db1", branch_source == 0 ? "t1$branch_dev" : "t1")) - .WithBranch(branch_source == 1 ? "dev" : "main") - .AddOption(Options::BRANCH, branch_source == 2 ? "dev" : "main"); - ASSERT_NOK_WITH_MSG(builder.Finish(), "requires the main branch"); + WriteContextBuilder option_only_builder("table_root_path", "commit_user_1"); + option_only_builder.WithCatalog(catalog, Identifier("db1", "t1")); + if (branch_source == 0) { + option_only_builder.WithBranch("dev"); + } else { + option_only_builder.AddOption(Options::BRANCH, "dev"); + } + ASSERT_NOK_WITH_MSG(option_only_builder.Finish(), + "name branch 'dev' there as 't1$branch_dev'"); } - WriteContextBuilder builder("table_root_path", "commit_user_1"); - ASSERT_OK(builder.WithCatalog(catalog, Identifier("db1", "t1$branch_main")) + + WriteContextBuilder mixed_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG(mixed_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_dev")) + .WithBranch("release") + .Finish(), + "but both 'dev' and 'release' were named"); + + WriteContextBuilder main_builder("table_root_path", "commit_user_1"); + ASSERT_OK(main_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_main")) .WithBranch("") .AddOption(Options::BRANCH, "main") .Finish()); + + WriteContextBuilder upper_main_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG( + upper_main_builder.WithCatalog(catalog, Identifier("db1", "t1$branch_MAIN")).Finish(), + "a catalog names branch 'MAIN' as it names the main branch"); + + WriteContextBuilder constructed_builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN( + auto constructed_ctx, + constructed_builder.WithCatalog(catalog, Identifier("db1", "t1", "dev")).Finish()); + ASSERT_EQ(constructed_ctx->GetBranch(), "dev"); + WriteContextBuilder system_table_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG( + system_table_builder.WithCatalog(catalog, Identifier("db1", "t1", "dev$options")).Finish(), + "Cannot 'write' for system table"); +} + +TEST(WriteContextTest, TestBranch) { + WriteContextBuilder option_builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN(auto option_ctx, option_builder.AddOption(Options::BRANCH, "rt").Finish()); + ASSERT_EQ("rt", option_ctx->GetBranch()); + + WriteContextBuilder upper_main_builder("table_root_path", "commit_user_1"); + ASSERT_OK_AND_ASSIGN(auto upper_main_ctx, upper_main_builder.WithBranch("MAIN").Finish()); + ASSERT_EQ("MAIN", upper_main_ctx->GetBranch()); + + // An empty branch selects the main branch and stays accepted. + WriteContextBuilder main_builder("table_root_path", "commit_user_1"); + main_builder.WithBranch(""); + ASSERT_OK_AND_ASSIGN(auto ctx, main_builder.Finish()); + ASSERT_EQ(BranchManager::DEFAULT_MAIN_BRANCH, ctx->GetBranch()); + WriteContextBuilder empty_builder("table_root_path", "commit_user_1"); + ASSERT_NOK_WITH_MSG(empty_builder.WithBranch("").AddOption(Options::BRANCH, "dev").Finish(), + "but both 'dev' and 'main' were named"); + + // The branch names a directory under the root path, so a value that is not a single path + // component is rejected when the context is built. + WriteContextBuilder escaping_builder("table_root_path", "commit_user_1"); + escaping_builder.WithBranch("rt/../../../../../outside"); + ASSERT_NOK_WITH_MSG(escaping_builder.Finish(), "branch name cannot contain path separators"); } TEST(WriteContextTest, TestSetContent) { @@ -137,20 +202,6 @@ TEST(WriteContextTest, TestSetOptionsOverridesAddedOptions) { ASSERT_EQ(expected_options, ctx->GetOptions()); } -TEST(WriteContextTest, TestRejectBranchLeavingRootPath) { - // The branch names a directory under the root path, so a value that is not a single path - // component is rejected when the context is built. - WriteContextBuilder builder("table_root_path", "commit_user_1"); - builder.WithBranch("rt/../../../../../outside"); - ASSERT_NOK_WITH_MSG(builder.Finish(), "branch name cannot contain path separators"); - - // An empty branch selects the main branch and stays accepted. - WriteContextBuilder main_builder("table_root_path", "commit_user_1"); - main_builder.WithBranch(""); - ASSERT_OK_AND_ASSIGN(auto ctx, main_builder.Finish()); - ASSERT_EQ("", ctx->GetBranch()); -} - TEST(WriteContextTest, TestSetWriteBufferSpillThreadNumber) { WriteContextBuilder builder("table_root_path", "commit_user_1"); builder.SetWriteBufferSpillThreadNumber(2); diff --git a/src/paimon/core/utils/branch_manager.cpp b/src/paimon/core/utils/branch_manager.cpp index 118189003..6a9a46b3d 100644 --- a/src/paimon/core/utils/branch_manager.cpp +++ b/src/paimon/core/utils/branch_manager.cpp @@ -20,14 +20,74 @@ #include "paimon/core/utils/branch_manager.h" #include +#include #include +#include #include +#include #include +#include "fmt/format.h" +#include "paimon/catalog/identifier.h" +#include "paimon/defs.h" #include "paimon/fs/file_system.h" namespace paimon { +Result BranchManager::ResolveBranch(const std::optional& identifier, + const std::map& options, + const std::optional& explicit_branch, + const std::string& operation) { + std::optional identifier_branch; + if (identifier) { + PAIMON_ASSIGN_OR_RAISE(identifier_branch, identifier.value().GetBranchName()); + } + auto option_iter = options.find(Options::BRANCH); + std::optional option_branch = + option_iter == options.end() ? std::nullopt + : std::optional(option_iter->second); + + std::optional resolved; + for (const std::optional& named_branch : + {identifier_branch, option_branch, explicit_branch}) { + if (!named_branch) { + continue; + } + std::string branch = NormalizeBranch(named_branch.value()); + // Checked before the comparison below, so that a name naming no directory of the table + // is reported as such rather than as one of two branches an operation was aimed at. + PAIMON_RETURN_NOT_OK(CheckValidBranch(branch)); + if (resolved && branch != resolved.value()) { + return Status::Invalid( + fmt::format("a {} is aimed at one branch, but both '{}' and '{}' were named", + operation, resolved.value(), branch)); + } + resolved = std::move(branch); + } + return resolved.value_or(DEFAULT_MAIN_BRANCH); +} + +Status BranchManager::CheckCatalogAddressableBranch(const std::string& branch) { + const std::string normalized = NormalizeBranch(branch); + if (!IsMainBranch(normalized) && + StringUtils::EqualsIgnoreCase(normalized, DEFAULT_MAIN_BRANCH)) { + return Status::Invalid(fmt::format( + "a catalog names branch '{}' as it names the main branch, while the snapshots of the " + "two go to directories of their own, so a commit to it through a catalog would publish " + "them on the main branch", + normalized)); + } + if (normalized.find(Identifier::kSystemTableSplitter) != std::string::npos) { + return Status::Invalid(fmt::format( + "a catalog names branch '{}' by the object name '{}{}{}', which reads back as " + "another branch or as a system table of one, so a branch a catalog addresses cannot " + "contain '{}'", + normalized, Identifier::kSystemTableSplitter, Identifier::kSystemBranchPrefix, + normalized, Identifier::kSystemTableSplitter)); + } + return Status::OK(); +} + Result> BranchManager::ListBranches(const std::shared_ptr& fs, const std::string& table_root) { std::vector branches = {DEFAULT_MAIN_BRANCH}; diff --git a/src/paimon/core/utils/branch_manager.h b/src/paimon/core/utils/branch_manager.h index 975cec9ae..eebb45647 100644 --- a/src/paimon/core/utils/branch_manager.h +++ b/src/paimon/core/utils/branch_manager.h @@ -17,10 +17,13 @@ */ #pragma once +#include #include +#include #include #include +#include "paimon/catalog/identifier.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/string_utils.h" #include "paimon/result.h" @@ -40,7 +43,7 @@ class BranchManager { static constexpr char DEFAULT_MAIN_BRANCH[] = "main"; static constexpr char BRANCH_PREFIX[] = "branch-"; - /// Normalizes an empty branch name to `main`. + /// Normalizes an empty or blank branch name to `main`. static std::string NormalizeBranch(const std::string& branch) { return StringUtils::IsNullOrWhitespaceOnly(branch) ? DEFAULT_MAIN_BRANCH : branch; } @@ -71,6 +74,28 @@ class BranchManager { return branch == DEFAULT_MAIN_BRANCH; } + /// Returns the branch an operation is aimed at, which the table identifier, the `branch` + /// option and `explicit_branch` may each name. + /// + /// An absent name and a name which `NormalizeBranch` maps to `main` both mean the main + /// branch. Names which disagree are refused rather than silently chosen between, with + /// `operation` naming the operation in that error. Every name is put through + /// `CheckValidBranch` before it is compared, so neither the branch this returns nor a + /// branch it names in an error is one that would leave the table root. + static Result ResolveBranch(const std::optional& identifier, + const std::map& options, + const std::optional& explicit_branch, + const std::string& operation); + + /// Fails when the branch object name `Identifier` builds, as `tbl$branch_dev`, does not read + /// back as `branch`, which is how a catalog is told which branch a commit is aimed at. + /// + /// That object name reads as the bare table for every spelling of `main`, while `BranchPath` + /// keeps a directory of its own for every spelling but `main`; and a branch holding a `$` + /// reads back as another branch, or as a system table of one. A branch this refuses can still + /// be committed to by writing the snapshot to its directory, which names no table. + static Status CheckCatalogAddressableBranch(const std::string& branch); + /// Lists all branches for a table, including `main`. static Result> ListBranches(const std::shared_ptr& fs, const std::string& table_root); diff --git a/src/paimon/core/utils/branch_manager_test.cpp b/src/paimon/core/utils/branch_manager_test.cpp index 74fdd047b..307a53f4b 100644 --- a/src/paimon/core/utils/branch_manager_test.cpp +++ b/src/paimon/core/utils/branch_manager_test.cpp @@ -18,7 +18,13 @@ #include "paimon/core/utils/branch_manager.h" +#include +#include +#include + #include "gtest/gtest.h" +#include "paimon/catalog/identifier.h" +#include "paimon/defs.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { @@ -47,6 +53,72 @@ TEST(BranchManagerTest, TestBranchPath) { ASSERT_EQ(BranchManager::BranchPath("/root", " "), "/root"); } +TEST(BranchManagerTest, TestResolveBranch) { + const std::map no_options; + ASSERT_OK_AND_ASSIGN(std::string unnamed, + BranchManager::ResolveBranch(/*identifier=*/std::nullopt, no_options, + /*explicit_branch=*/std::nullopt, "commit")); + ASSERT_EQ(BranchManager::DEFAULT_MAIN_BRANCH, unnamed); + + ASSERT_OK_AND_ASSIGN(std::string from_identifier, + BranchManager::ResolveBranch(Identifier("db", "tbl$branch_data"), + no_options, std::nullopt, "commit")); + ASSERT_EQ("data", from_identifier); + ASSERT_OK_AND_ASSIGN(std::string from_option, + BranchManager::ResolveBranch(std::nullopt, {{Options::BRANCH, "data"}}, + std::nullopt, "commit")); + ASSERT_EQ("data", from_option); + ASSERT_OK_AND_ASSIGN(std::string from_given, + BranchManager::ResolveBranch(std::nullopt, no_options, "data", "write")); + ASSERT_EQ("data", from_given); + + ASSERT_OK_AND_ASSIGN(std::string empty, + BranchManager::ResolveBranch(std::nullopt, no_options, "", "write")); + ASSERT_EQ(BranchManager::DEFAULT_MAIN_BRANCH, empty); + ASSERT_OK_AND_ASSIGN(std::string agreed, BranchManager::ResolveBranch( + Identifier("db", "tbl$branch_data"), + {{Options::BRANCH, "data"}}, "data", "write")); + ASSERT_EQ("data", agreed); + + ASSERT_NOK_WITH_MSG( + BranchManager::ResolveBranch(Identifier("db", "tbl$branch_data"), {{Options::BRANCH, "rt"}}, + std::nullopt, "commit"), + "a commit is aimed at one branch, but both 'data' and 'rt' were named"); + ASSERT_NOK_WITH_MSG( + BranchManager::ResolveBranch(std::nullopt, {{Options::BRANCH, "main"}}, "rt", "write"), + "a write is aimed at one branch, but both 'main' and 'rt' were named"); + + ASSERT_NOK_WITH_MSG(BranchManager::ResolveBranch(Identifier("db", "tbl$a$b$c"), no_options, + std::nullopt, "commit"), + "Invalid table name"); + + ASSERT_NOK_WITH_MSG( + BranchManager::ResolveBranch(Identifier("db", "tbl$branch_data"), + {{Options::BRANCH, "rt/../../outside"}}, std::nullopt, "read"), + "branch name cannot contain path separators"); + ASSERT_NOK_WITH_MSG( + BranchManager::ResolveBranch(std::nullopt, no_options, "line\nfeed", "write"), + "branch name cannot contain control characters"); +} + +TEST(BranchManagerTest, TestCheckCatalogAddressableBranch) { + ASSERT_OK(BranchManager::CheckCatalogAddressableBranch(BranchManager::DEFAULT_MAIN_BRANCH)); + ASSERT_OK(BranchManager::CheckCatalogAddressableBranch("")); + ASSERT_OK(BranchManager::CheckCatalogAddressableBranch(" ")); + ASSERT_OK(BranchManager::CheckCatalogAddressableBranch("data")); + ASSERT_OK(BranchManager::CheckCatalogAddressableBranch("mainline")); + + ASSERT_NOK_WITH_MSG(BranchManager::CheckCatalogAddressableBranch("MAIN"), + "a catalog names branch 'MAIN' as it names the main branch"); + ASSERT_NOK_WITH_MSG(BranchManager::CheckCatalogAddressableBranch("Main"), + "a catalog names branch 'Main' as it names the main branch"); + + ASSERT_NOK_WITH_MSG(BranchManager::CheckCatalogAddressableBranch("dev$options"), + "a branch a catalog addresses cannot contain '$'"); + ASSERT_NOK_WITH_MSG(BranchManager::CheckCatalogAddressableBranch("a$b$c"), + "a branch a catalog addresses cannot contain '$'"); +} + TEST(BranchManagerTest, TestCheckValidBranch) { ASSERT_OK(BranchManager::CheckValidBranch(BranchManager::DEFAULT_MAIN_BRANCH)); ASSERT_OK(BranchManager::CheckValidBranch("data")); diff --git a/src/paimon/rest/rest_catalog.cpp b/src/paimon/rest/rest_catalog.cpp index 953263764..64396a297 100644 --- a/src/paimon/rest/rest_catalog.cpp +++ b/src/paimon/rest/rest_catalog.cpp @@ -60,27 +60,14 @@ std::optional NormalizeBranch(std::optional branch) { return branch; } -// Builds the "
$branch_" object name addressing `branch` of `table_name` -// on the rest server. -std::string BranchObjectName(std::string table_name, const std::string& branch) { - table_name.append(Identifier::kSystemTableSplitter); - table_name.append(Identifier::kSystemBranchPrefix); - table_name.append(branch); - return table_name; -} - // Builds the identifier sent to the rest server: the system table suffix is stripped // while the branch stays in the object name, so the server resolves the branch itself and // returns the branch's own schema. The path the server reports is the data table root in // either case; the branch subdirectory is derived downstream from the branch option and // must not be applied twice (see `ToTableSchema`). Result ToLoadIdentifier(const Identifier& identifier) { - PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); PAIMON_ASSIGN_OR_RAISE(std::optional branch, identifier.GetBranchName()); - branch = NormalizeBranch(std::move(branch)); - std::string object_name = branch ? BranchObjectName(std::move(data_table_name), branch.value()) - : std::move(data_table_name); - return Identifier(identifier.GetDatabaseName(), object_name); + return CatalogUtils::BranchIdentifier(identifier, branch.value_or(std::string())); } } // namespace @@ -455,12 +442,9 @@ Result> RestCatalog::ListSnapshots(const Identifier& i const std::string& branch) const { PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotBranch(identifier, "listSnapshots")); PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotSystemTable(identifier, "listSnapshots")); - std::optional normalized_branch = - NormalizeBranch(branch.empty() ? std::nullopt : std::make_optional(branch)); - std::string object_name = - normalized_branch ? BranchObjectName(identifier.GetTableName(), normalized_branch.value()) - : identifier.GetTableName(); - Identifier load_identifier(identifier.GetDatabaseName(), object_name); + // `CheckNotBranch` and `CheckNotSystemTable` above leave the table name bare, so the branch + // this is asked for is the only one the object name can carry. + Identifier load_identifier(identifier.GetDatabaseName(), identifier.GetTableName(), branch); // The Catalog interface has no pagination, so all pages are fetched; the server // does not order snapshots across pages while the contract requires ascending ids. PAIMON_ASSIGN_OR_RAISE(std::vector snapshots, api_->ListSnapshots(load_identifier)); diff --git a/src/paimon/rest/rest_catalog_test.cpp b/src/paimon/rest/rest_catalog_test.cpp index d48bd2adf..197877e25 100644 --- a/src/paimon/rest/rest_catalog_test.cpp +++ b/src/paimon/rest/rest_catalog_test.cpp @@ -1262,6 +1262,10 @@ TEST_F(RestCatalogTest, ListSnapshots) { ASSERT_OK_AND_ASSIGN(std::vector main_case_snapshots, catalog->ListSnapshots(identifier, "MAIN")); ASSERT_EQ(2, main_case_snapshots.size()); + // a blank name is the main branch everywhere in this library, so it is the bare table here + ASSERT_OK_AND_ASSIGN(std::vector blank_snapshots, + catalog->ListSnapshots(identifier, " ")); + ASSERT_EQ(2, blank_snapshots.size()); Status missing = catalog->ListSnapshots(Identifier("db1", "t9"), "").status(); ASSERT_TRUE(missing.IsNotExist()) << missing.ToString(); diff --git a/src/paimon/testing/mock/mock_catalog.h b/src/paimon/testing/mock/mock_catalog.h index 0931ba799..3118cfa7e 100644 --- a/src/paimon/testing/mock/mock_catalog.h +++ b/src/paimon/testing/mock/mock_catalog.h @@ -104,15 +104,19 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog { } Result> LoadTableSchema(const Identifier& identifier) const override { - ++load_table_schema_calls_; + load_table_schema_identifiers_.push_back(identifier); if (table_schema_ == nullptr) { return Status::NotExist(identifier.ToString() + " not exist"); } return table_schema_; } + const std::vector& LoadTableSchemaIdentifiers() const { + return load_table_schema_identifiers_; + } + size_t LoadTableSchemaCalls() const { - return load_table_schema_calls_; + return load_table_schema_identifiers_.size(); } Result> LoadSnapshot(const Identifier& identifier) const override { @@ -248,7 +252,7 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog { std::optional current_snapshot_uuid_; std::vector accepted_; mutable size_t get_table_calls_ = 0; - mutable size_t load_table_schema_calls_ = 0; + mutable std::vector load_table_schema_identifiers_; mutable std::vector load_snapshot_identifiers_; bool serve_snapshots_ = false; Status load_snapshot_status_; diff --git a/test/inte/pk_compaction_inte_test.cpp b/test/inte/pk_compaction_inte_test.cpp index 0ec5b563b..7e713b067 100644 --- a/test/inte/pk_compaction_inte_test.cpp +++ b/test/inte/pk_compaction_inte_test.cpp @@ -1882,9 +1882,9 @@ TEST_F(PkCompactionInteTest, DeduplicateWithRowKindAndDV) { } } -// Test: write and compact on branch. -// Copies append_table_with_rt_branch.db to a temp dir, writes data to branch-rt, then compacts. -// Since we don't support branch commit and scan, we only verify the commit messages. +// Test: write, compact and commit on branch. +// Copies append_table_with_rt_branch.db to a temp dir, writes data to branch-rt, compacts, commits +// to that branch and scans it back. TEST_F(PkCompactionInteTest, WriteAndCompactWithBranch) { // Step 1: Copy the table with branch to a temp directory. std::string src_table_path = @@ -1954,35 +1954,78 @@ 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 commit_context, commit_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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(fake_split))); - ASSERT_OK_AND_ASSIGN(auto result_array, - ReadResultCollector::CollectResult(std::move(batch_reader))); + auto fs = std::make_shared(); + 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 and read branch-rt. + std::map 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 scan_context, scan_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(scan_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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> splits = plan->Splits(); + ASSERT_FALSE(splits.empty()); + for (const std::shared_ptr& split : splits) { + auto* split_impl = dynamic_cast(split.get()); + ASSERT_NE(split_impl, nullptr); + ASSERT_EQ(split_impl->SnapshotId(), 3); + } + + ReadContextBuilder read_context_builder(table_path); + read_context_builder.SetOptions(branch_options); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_context, read_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_read, + TableRead::Create(std::move(read_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_reader, + table_read->CreateReader(splits)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr result_array, + ReadResultCollector::CollectResult(std::move(batch_reader))); + ASSERT_TRUE(result_array); - std::shared_ptr expected_array; - auto status = arrow::ipc::internal::json::ChunkedArrayFromJSON(result_type, {R"([ + 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::shared_ptr expected_array; + auto status = arrow::ipc::internal::json::ChunkedArrayFromJSON(result_type, {R"([ +[0, "20240725", "apple", 4], +[0, "20240725", "peach", 10], [0, "20240726", "cherry", 30], -[0, "20240726", "grape", 40] +[0, "20240726", "grape", 40], +[0, "20240726", "pear", 6] ])"}, - &expected_array); - ASSERT_TRUE(status.ok()); - ASSERT_TRUE(result_array); - ASSERT_TRUE(result_array->Equals(*expected_array)); - } + &expected_array); + ASSERT_TRUE(status.ok()); + auto result_schema = std::make_shared(fields_with_row_kind); + ASSERT_OK_AND_ASSIGN(expected_array, + ReadResultCollector::SortArray(expected_array, result_schema)); + ASSERT_OK_AND_ASSIGN(result_array, ReadResultCollector::SortArray(result_array, result_schema)); + ASSERT_TRUE(result_array->Equals(*expected_array)) << result_array->ToString(); } TEST_F(PkCompactionInteTest, TestPartialUpdateNoDv) { diff --git a/test/inte/write_inte_test.cpp b/test/inte/write_inte_test.cpp index c7848ecdc..31a325ec1 100644 --- a/test/inte/write_inte_test.cpp +++ b/test/inte/write_inte_test.cpp @@ -3619,6 +3619,74 @@ TEST_F(WriteInteTest, TestBranchWrite) { ASSERT_TRUE(expected_commit_message_1->TEST_Equal(*result_commit_msgs1)); } +TEST_F(WriteInteTest, TestBranchCommitAndRead) { + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::int32())}; + auto schema = arrow::schema(fields); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::map options = {{Options::FILE_FORMAT, "parquet"}, + {Options::FILE_SYSTEM, "local"}, + {Options::BUCKET, "-1"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr main_helper, + TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{}, + /*primary_keys=*/{}, options, + /*is_streaming_mode=*/false)); + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr main_batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([["main", 1]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(main_helper->WriteAndCommit(std::move(main_batch), /*commit_identifier=*/0, + /*expected_commit_messages=*/std::nullopt)); + + SchemaManager branch_schema_manager(file_system_, table_path, "dev"); + ASSERT_OK(branch_schema_manager.CreateTable(schema, /*partition_keys=*/{}, + /*primary_keys=*/{}, options)); + + std::map branch_options = options; + branch_options[Options::BRANCH] = "dev"; + ASSERT_OK_AND_ASSIGN( + std::unique_ptr branch_helper, + TestHelper::Create(table_path, branch_options, /*is_streaming_mode=*/false)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr branch_batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([["dev", 2]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(branch_helper->WriteAndCommit(std::move(branch_batch), /*commit_identifier=*/1, + /*expected_commit_messages=*/std::nullopt)); + + std::string branch_snapshots = PathUtil::JoinPath(table_path, "branch/branch-dev/snapshot"); + ASSERT_OK_AND_ASSIGN(bool branch_snapshot_exists, + file_system_->Exists(PathUtil::JoinPath(branch_snapshots, "snapshot-1"))); + ASSERT_TRUE(branch_snapshot_exists); + ASSERT_OK_AND_ASSIGN(bool branch_hint_exists, + file_system_->Exists(PathUtil::JoinPath(branch_snapshots, "LATEST"))); + ASSERT_TRUE(branch_hint_exists); + ASSERT_OK_AND_ASSIGN(bool main_snapshot_exists, file_system_->Exists(PathUtil::JoinPath( + table_path, "snapshot/snapshot-1"))); + ASSERT_TRUE(main_snapshot_exists); + ASSERT_OK_AND_ASSIGN(bool main_holds_branch_snapshot, file_system_->Exists(PathUtil::JoinPath( + table_path, "snapshot/snapshot-2"))); + ASSERT_FALSE(main_holds_branch_snapshot); + + arrow::FieldVector fields_with_row_kind = fields; + fields_with_row_kind.insert(fields_with_row_kind.begin(), + arrow::field("_VALUE_KIND", arrow::int8())); + auto data_type = arrow::struct_(fields_with_row_kind); + ASSERT_OK_AND_ASSIGN( + std::vector> branch_splits, + branch_helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt)); + ASSERT_OK_AND_ASSIGN(bool branch_read, branch_helper->ReadAndCheckResult( + data_type, branch_splits, R"([[0, "dev", 2]])")); + ASSERT_TRUE(branch_read); + + ASSERT_OK_AND_ASSIGN( + std::vector> main_splits, + main_helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt)); + ASSERT_OK_AND_ASSIGN(bool main_read, main_helper->ReadAndCheckResult(data_type, main_splits, + R"([[0, "main", 1]])")); + ASSERT_TRUE(main_read); +} + TEST_P(WriteInteTest, TestDataEvolutionWrite) { arrow::FieldVector fields = { arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int8()),