feat(catalog): support writing and committing to a branch

A commit was hard-wired to the main branch: `FileStoreCommit::Create()`
built its schema and snapshot managers without a branch, so a commit
configured with `branch=dev` read the main schema and published its
snapshot under the main branch's directory. A write through a catalog
refused a branch outright.

Resolve the branch an operation is aimed at from every source that may
name it - the table identifier, the `branch` option and, where a builder
has one, `WithBranch()` - and refuse names that disagree rather than
choosing between them. A read resolves it the same way, so the `branch`
option now names a read's branch as `WithBranch()` already did.

A catalog is told which branch to answer for by the
`<table>$branch_<branch>` object name that `Identifier(db, table,
branch)` builds and `CatalogUtils::BranchIdentifier()` rebuilds for
every catalog call, as the Java client does. `SnapshotCommit::Commit()`
now takes that branch, so a renaming commit writes through a manager of
the branch and a catalog commit addresses the branch's object name.
`UseRESTCatalogCommit()` still builds a body naming no branch, so the
caller sends it to the branch's own URL.

A branch keeps its schema, snapshots and real-time offsets under
`branch/branch-<name>`, while data and manifests stay shared under the
table path. A write or a commit aimed at a branch reads the schema
published there rather than the catalog's, as
`AbstractFileStoreTable.switchToBranch()` does, so a catalog serving
only the table's own current schema still takes the branch's snapshots;
only the main branch reloads the catalog's schema id for a new snapshot.

Refuse the two kinds of name a catalog cannot tell apart from another
object: a branch differing from `main` by case alone, whose snapshots
would go to a directory of their own, and a branch holding a `$`, which
reads back as a system table of another branch.

Expiration and snapshot loading follow the commit's branch. Expiration
reads the retained snapshots of that branch alone, as Java does, while
every branch shares the table's data files, so a file that only another
branch still refers to is deleted, which the public header and the
catalog guide both state.

Author: SteNicholas

docs/source/user_guide/catalog.rst

@@ -180,11 +180,8 @@ 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 keeps both under its own directory, 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 +190,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 +222,73 @@ restored by ``RollbackToAsLatest``. Serialize rollback and expiration through
 the upstream coordinator: a rollback must finish before expiration starts,
 so its restored file references are visible to the expiration operation.
 
+.. warning::
+
+   Expiration reads the retained snapshots of the branch it runs on and of no
+   other, while data files are shared by every branch of the table, so a file
+   that only another branch still refers to is not preserved by that reference
+   and is deleted. This holds for the main branch as much as for the others:
+   expiring the main branch deletes a file only a branch refers to. Expire only
+   where the retained snapshots of the branch cover every file the other branches
+   still read, or keep those files reachable from the expiring branch through the
+   upstream coordinator.
+
 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-<name>``, 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 keeps its schema, snapshots and real-time
+offsets under ``branch/branch-<name>``; the main branch keeps them under the
+table path. Data and manifests are shared by all branches under the table path,
+which is what the expiration warning above turns on.
+
+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-<name>``; 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.

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-<name>`` 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.

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;

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>& catalog,
                                       const Identifier& identifier);

include/paimon/file_store_commit.h

@@ -164,7 +164,9 @@ class PAIMON_EXPORT FileStoreCommit {
         const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
         int64_t commit_identifier, std::optional<int64_t> 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,9 @@ 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 Only the retained snapshots of the branch this commit was built for are read, while
+    ///       the branches of a table share their data files, so a file that only another branch
+    ///       still refers to is deleted. This holds for the main branch as much as for the others.
     /// @return Result<int32_t> indicating the number of expired items or an error status.
     virtual Result<int32_t> Expire() = 0;
 

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

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.

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<bool> Commit(const std::optional<std::string>& base_snapshot_uuid,
-                        const Snapshot& snapshot,
+                        const Snapshot& snapshot, const std::string& branch,
                         const std::vector<PartitionStatistics>& statistics) override {
+        // 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<std::string> 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.");
         }
     }