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

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-<name>`` 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-<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 publishes its schema and real-time offsets
+under ``branch/branch-<name>``, 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-<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/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-<name>``, 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
 ~~~~~~~~~~~~~~

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,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-<name>`: 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<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.");
         }
     }