Skip to content

Commit 2711cf2

Browse files
authored
feat(catalog): support writing and committing to a branch (#357)
1 parent b8b8cfe commit 2711cf2

37 files changed

Lines changed: 1329 additions & 248 deletions

‎docs/source/user_guide/catalog.rst‎

Lines changed: 75 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -180,11 +180,9 @@ Its metadata files are retained. Use ``FilterAndCommit`` for batch recovery;
180180
after a real-time error, recreate the writer and context and recover from durable
181181
offsets as described by ``CommitWithProgress``.
182182

183-
Only the main branch is supported by ``WithCatalog``. Both builders reject other
184-
branches in the identifier or ``branch`` option; the write builder also checks
185-
``WithBranch``. Explicit ``tbl$branch_main`` and ``branch=main`` are accepted.
186-
187-
The current schema and latest snapshot need no files under the table path.
183+
The main branch's current schema and latest snapshot need no files under the table
184+
path. A branch needs its schema published under its own directory, while its
185+
latest snapshot can be held by the catalog alone, as described below.
188186
A catalog response of ``{"snapshot": null}`` means the table has no snapshot.
189187
Only a catalog reporting that snapshot loading is unsupported falls back to
190188
file-system lookup; other catalog errors propagate to the caller.
@@ -193,7 +191,8 @@ Read ``Table::CatalogUuid()`` before preparing changes and pass it to
193191
``WithTableId`` so the server can reject commits to a dropped and recreated table.
194192
``Table::Uuid()`` can fall back to the table name and must not supply this id.
195193
An unset table id is sent as null and subject to server validation. Each commit
196-
attempt reloads the catalog's current schema id for the new snapshot.
194+
attempt to the main branch reloads the catalog's current schema id for the new
195+
snapshot; on a branch it reloads the id published under the branch.
197196

198197
.. note::
199198

@@ -224,7 +223,76 @@ restored by ``RollbackToAsLatest``. Serialize rollback and expiration through
224223
the upstream coordinator: a rollback must finish before expiration starts,
225224
so its restored file references are visible to the expiration operation.
226225

226+
.. warning::
227+
228+
``Expire`` returns ``NotImplemented`` before reading any snapshot or deleting
229+
any file when it runs on a branch other than main, or finds one under
230+
``branch/branch-<name>`` of the table path. Data files are shared by every
231+
branch of the table, while expiration reads only the retained snapshots of the
232+
branch it runs on, so it could delete a file another branch still refers to.
233+
A branch the catalog holds without that directory, or one created while
234+
expiration runs, is not found: do not expire a table with such a branch, and
235+
serialize branch creation and expiration through the upstream coordinator.
236+
227237
The C++ REST catalog covers the database, table, snapshot and commit operations
228238
of the ``Catalog`` API. The parts of the Java REST catalog that have no C++
229239
counterpart yet — altering a database or a table, views, functions, partitions,
230-
tags, branch management and consumers — are not supported.
240+
tags, branch management and consumers — are not supported. Creating, deleting
241+
and merging a branch is what branch management covers; committing to a branch
242+
that already exists is supported, as described below.
243+
244+
Committing to a branch
245+
~~~~~~~~~~~~~~~~~~~~~~
246+
This library creates no branch: one has to exist, with its schema published under
247+
``branch/branch-<name>``, before a write or a commit can be aimed at it.
248+
249+
Naming the branch
250+
^^^^^^^^^^^^^^^^^
251+
A branch is addressed by the identifier of ``WithCatalog``, which
252+
``paimon::Identifier("db", "tbl", "dev")`` builds as ``tbl$branch_dev``, so that
253+
the catalog answers for the branch: the snapshot it loads is the branch's, and
254+
the commit request is addressed to that branch by that identifier rather than by
255+
anything in its body. The ``branch`` option and, for writers, ``WithBranch`` may
256+
name the same branch as well; naming two different branches is rejected rather
257+
than resolved, and ``tbl$branch_main``, ``branch=main`` and an empty name all
258+
mean the main branch. Naming a branch only in the option or ``WithBranch`` while
259+
the identifier names the bare table is rejected, as the catalog would then answer
260+
for the table while the files were written for the branch.
261+
262+
Names a catalog cannot tell apart from another object are rejected where a
263+
builder sees them: a name differing from ``main`` by case alone, such as
264+
``tbl$branch_MAIN`` or ``branch=MAIN``, which a catalog reads as the main branch
265+
while its snapshots go to a directory of their own; and a name holding a ``$``,
266+
such as ``branch=dev$options``, which ``tbl$branch_dev$options`` reads back as
267+
the ``options`` system table of branch ``dev``. ``paimon::Identifier("db",
268+
"tbl", "MAIN")`` never reaches that check: the constructor folds every spelling
269+
of ``main`` into the bare table name, as the Java client does, and so names the
270+
main branch.
271+
272+
What a branch keeps and what it shares
273+
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
274+
A branch other than the main one publishes its schema and real-time offsets
275+
under ``branch/branch-<name>``, where a snapshot written to the file system goes
276+
as well; the main branch keeps these files under the table path. A
277+
version-managed catalog can hold the latest snapshot of either branch alone.
278+
Data and manifests are shared by all branches under the table path,
279+
which is why expiration is not supported on a table with branches, as noted
280+
above.
281+
282+
A write or a commit aimed at a branch reads that schema rather than the
283+
catalog's, just as a read of that branch does, and asks the catalog for no schema
284+
at all, so a catalog serving only the table's own current schema still takes the
285+
branch's snapshots. The new snapshot records the schema id published there as
286+
well, which is the id a read of the branch resolves under
287+
``branch/branch-<name>``; only on the main branch is the recorded id the one the
288+
catalog reports, as its current schema may live nowhere else. Data files record
289+
the schema they were written with, as they do on the main branch.
290+
291+
Addressing a branch without a catalog client
292+
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
293+
``UseRESTCatalogCommit`` builds the request instead of sending it, and takes the
294+
branch from the ``branch`` option alone, as it is configured without an
295+
identifier. The request body carries the table id but no table name and no
296+
branch, so the branch has to appear in the URL the caller sends that body to:
297+
the commit endpoint of ``tbl$branch_dev``, not the one of ``tbl``. A body sent
298+
to the bare table's URL publishes the branch's snapshot on the main branch.

‎docs/source/user_guide/clean.rst‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,13 @@ Description
113113

114114
- The snapshot expiration interface deletes data files according to the
115115
expiration policy and returns the number of snapshots deleted.
116+
- Expiration on a branch other than main, or on a table with such a branch under
117+
``branch/branch-<name>``, is not supported: branches share the table's data
118+
files, so expiration returns ``NotImplemented`` before deleting anything.
119+
- A branch is found only by that directory. A branch held only by a catalog, or
120+
one created while expiration runs, is not found, and expiration could delete
121+
data files it still refers to: do not expire a table with such a branch, and
122+
serialize branch creation and expiration.
116123

117124
Detailed Steps
118125
~~~~~~~~~~~~~~

‎docs/source/user_guide/write.rst‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -172,14 +172,16 @@ Operational Flow
172172
For a table of a catalog which owns its versions, such as a REST catalog, it
173173
passes that catalog and the table identifier to
174174
``WriteContextBuilder::WithCatalog`` and ``CommitContextBuilder::WithCatalog``.
175-
The writer loads the current schema and latest snapshot from the catalog.
176-
On ``Commit``, the snapshot and
177-
the statistics of the change go to the catalog, which decides whether this
178-
commit wins, and one that lost the race is rebased and retried the way a
179-
file-system commit is. A caller without a catalog client can instead enable
180-
REST catalog commit mode, call ``Commit``, obtain the JSON request from
181-
``GetLastCommitTableRequest``, and send that request to the REST catalog
182-
itself.
175+
The writer loads the latest snapshot from the catalog, and the current schema
176+
from it as well when it writes to the main branch; a write aimed at another
177+
branch reads the schema published under ``branch/branch-<name>`` instead. On
178+
``Commit``, the snapshot and the statistics of the change go to the catalog,
179+
which decides whether this commit wins, and one that lost the race is rebased
180+
and retried the way a file-system commit is. A caller without a catalog
181+
client can instead enable REST catalog commit mode, call ``Commit``, obtain
182+
the JSON request from ``GetLastCommitTableRequest``, and send that request to
183+
the REST catalog itself. That request body names no branch, so a commit aimed
184+
at one has to be sent to the URL of ``tbl$branch_dev`` rather than of ``tbl``.
183185

184186
6. The local committer or REST catalog validates the messages, updates
185187
manifests/metadata, and finalizes the snapshot atomically.

‎include/paimon/catalog/identifier.h‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,19 @@ class PAIMON_EXPORT Identifier {
3737

3838
explicit Identifier(const std::string& table);
3939
Identifier(const std::string& database, const std::string& table);
40+
/// Names `branch` of `table`: every branch but the main one is addressed by the object name
41+
/// this builds, which joins them as `tbl$branch_dev`.
42+
///
43+
/// The branch is matched against the main branch ignoring case, as the Java client matches it,
44+
/// so `MAIN` names the main branch here rather than a branch of its own. A branch of only
45+
/// whitespace names it too, which the Java client does not do: this library reads a blank
46+
/// branch as the main branch everywhere.
47+
///
48+
/// @param database Database holding the table.
49+
/// @param table Data table name, carrying no branch or system table suffix.
50+
/// @param branch Branch of `table`; any spelling of `main`, an empty and a blank name all
51+
/// name the main branch.
52+
Identifier(const std::string& database, const std::string& table, const std::string& branch);
4053

4154
bool operator==(const Identifier& other) const;
4255
const std::string& GetDatabaseName() const;

‎include/paimon/commit_context.h‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -195,8 +195,11 @@ class PAIMON_EXPORT CommitContextBuilder {
195195
/// `Expire()` manages file-system snapshots after confirming that the catalog's current
196196
/// snapshot and retained history are published there. Configure writers with the same catalog
197197
/// via `WriteContextBuilder::WithCatalog()`.
198+
/// A branch is addressed by the identifier, as `tbl$branch_dev`, so that the catalog answers
199+
/// for that branch. Its schema is read from the branch directory rather than from the catalog,
200+
/// just as a read of that branch reads it. Configure the writer with the same identifier.
198201
/// @param catalog Non-null catalog, kept alive by this context.
199-
/// @param identifier Table to commit to. Only the main branch is supported.
202+
/// @param identifier Table to commit to, naming a branch of it as `tbl$branch_dev`.
200203
/// @return Reference to this builder for method chaining.
201204
CommitContextBuilder& WithCatalog(const std::shared_ptr<Catalog>& catalog,
202205
const Identifier& identifier);

‎include/paimon/file_store_commit.h‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -164,7 +164,9 @@ class PAIMON_EXPORT FileStoreCommit {
164164
const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
165165
int64_t commit_identifier, std::optional<int64_t> watermark = std::nullopt) = 0;
166166

167-
/// Returns the request from the latest commit attempt, including failed attempts.
167+
/// Returns the request from the latest commit attempt, including failed attempts. Each attempt
168+
/// clears the request of the one before it, so an attempt that failed before building one, as
169+
/// one naming a branch a catalog cannot address does, leaves this returning an error.
168170
/// Catalog commits send requests automatically; `UseRESTCatalogCommit()` only prepares them.
169171
/// @note Temporary interface for internal use, will be removed in the future.
170172
/// @return JSON with `tableId`, `baseSnapshotUuid`, `snapshot` and `statistics`.
@@ -179,6 +181,12 @@ class PAIMON_EXPORT FileStoreCommit {
179181
/// catalog or retained-metadata read errors propagate without deleting files.
180182
/// Retained snapshots with index manifests return `NotImplemented` before deletion.
181183
/// @note Coordinate rollback and expiration so they do not run concurrently.
184+
/// @note Returns `NotImplemented` before any snapshot is read or any file is deleted when this
185+
/// commit is on a branch other than main, or the table path holds such a branch under
186+
/// `branch/branch-<name>`: branches share the table's data files, while expiration only
187+
/// reads the retained snapshots of its own branch. A branch held only by a catalog, or
188+
/// created while expiration runs, is not found; do not expire a table with such a
189+
/// branch, and serialize branch creation and expiration.
182190
/// @return Result<int32_t> indicating the number of expired items or an error status.
183191
virtual Result<int32_t> Expire() = 0;
184192

‎include/paimon/read_context.h‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -443,9 +443,12 @@ class PAIMON_EXPORT ReadContextBuilder {
443443
/// Paimon supports branching for data versioning and time travel queries.
444444
/// This method allows reading from a specific branch instead of the main branch.
445445
///
446+
/// The `branch` option names the branch too, as it does for a scan of one branch; naming two
447+
/// different branches is refused rather than silently resolved.
448+
///
446449
/// @param branch Name of the branch to read from.
447450
/// @return Reference to this builder for method chaining.
448-
/// @note Default branch is "main" if not specified.
451+
/// @note Default branch is "main" if not specified. An empty name is the main branch.
449452
ReadContextBuilder& WithBranch(const std::string& branch);
450453

451454
/// Sets a mapping from URI schemes (e.g., "file", "oss") to registered file system

‎include/paimon/write_context.h‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -238,6 +238,11 @@ class PAIMON_EXPORT WriteContextBuilder {
238238
WriteContextBuilder& WithWriteId(int32_t write_id);
239239

240240
/// Write to specific branch, default is main.
241+
///
242+
/// The identifier of `WithCatalog()` and the `branch` option may name the branch too; naming
243+
/// two different ones is refused rather than silently resolved. An empty name is the main
244+
/// branch.
245+
/// @param branch Name of the branch to write to.
241246
/// @return Reference to this builder for method chaining.
242247
WriteContextBuilder& WithBranch(const std::string& branch);
243248

@@ -258,9 +263,12 @@ class PAIMON_EXPORT WriteContextBuilder {
258263
/// snapshot. Use the same catalog and identifier as `CommitContextBuilder` to restore and
259264
/// refresh real-time progress.
260265
///
261-
/// Only the main branch is supported. Data, manifests and historical metadata remain on the
262-
/// file system. The catalog supplies that file system unless overridden
263-
/// by `WithFileSystem()` or `WithFileSystemSchemeToIdentifierMap()`.
266+
/// A branch is addressed by the identifier, as `tbl$branch_dev`, so that the catalog answers
267+
/// for that branch. Its schema is read from the branch directory rather than from the catalog,
268+
/// just as a read of that branch reads it.
269+
/// Data, manifests and historical metadata remain on the file system. The catalog supplies
270+
/// that file system unless overridden by `WithFileSystem()` or
271+
/// `WithFileSystemSchemeToIdentifierMap()`.
264272
/// @param catalog Non-null catalog, kept alive while the writer uses its snapshot loader.
265273
/// @param identifier The native table to write to.
266274
/// @return Reference to this builder for method chaining.

‎src/paimon/core/catalog/catalog_snapshot_commit.h‎

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,13 @@
2626

2727
#include "paimon/catalog/catalog.h"
2828
#include "paimon/catalog/identifier.h"
29+
#include "paimon/core/catalog/catalog_utils.h"
2930
#include "paimon/core/catalog/commit_table_request.h"
3031
#include "paimon/core/catalog/snapshot_commit.h"
3132
#include "paimon/core/catalog/version_managed_catalog.h"
33+
#include "paimon/core/utils/branch_manager.h"
34+
#include "paimon/result.h"
35+
#include "paimon/status.h"
3236

3337
namespace paimon {
3438

@@ -53,8 +57,14 @@ class CatalogSnapshotCommit : public SnapshotCommit {
5357
: identifier_("", ""), table_id_(table_id) {}
5458

5559
Result<bool> Commit(const std::optional<std::string>& base_snapshot_uuid,
56-
const Snapshot& snapshot,
60+
const Snapshot& snapshot, const std::string& branch,
5761
const std::vector<PartitionStatistics>& statistics) override {
62+
// Dropped before anything can refuse this attempt, so that what a caller reads back is
63+
// the attempt it just made or nothing at all, never the snapshot of an earlier one.
64+
commit_table_request_.reset();
65+
// Refused before the request is built: a branch whose object name reads back as another
66+
// object is one that neither this nor the caller of a request can commit to.
67+
PAIMON_RETURN_NOT_OK(BranchManager::CheckCatalogAddressableBranch(branch));
5868
commit_table_request_ =
5969
CommitTableRequest(table_id_, base_snapshot_uuid, snapshot, statistics);
6070
if (catalog_ == nullptr) {
@@ -65,8 +75,11 @@ class CatalogSnapshotCommit : public SnapshotCommit {
6575
"this catalog does not manage the versions of its tables, so there is nothing to "
6676
"commit the snapshot to");
6777
}
68-
return version_managed_catalog_->CommitSnapshot(identifier_, table_id_, base_snapshot_uuid,
69-
snapshot, statistics);
78+
// The branch this commit is aimed at wins over the one the identifier was built for.
79+
PAIMON_ASSIGN_OR_RAISE(Identifier branch_identifier,
80+
CatalogUtils::BranchIdentifier(identifier_, branch));
81+
return version_managed_catalog_->CommitSnapshot(branch_identifier, table_id_,
82+
base_snapshot_uuid, snapshot, statistics);
7083
}
7184

7285
std::string DescribeTarget() const override {
@@ -78,12 +91,16 @@ class CatalogSnapshotCommit : public SnapshotCommit {
7891
return catalog_ == nullptr;
7992
}
8093

81-
/// Returns the request from the latest attempt, including refusals and errors.
94+
/// Returns the request from the latest attempt, including refusals and errors. An attempt
95+
/// refused before the request is built, as one naming an unaddressable branch is, leaves no
96+
/// request behind rather than the one of the attempt before it.
8297
Result<std::string> GetLastCommitTableRequest() override {
8398
if (commit_table_request_) {
8499
return commit_table_request_.value().ToJsonString();
85100
} else {
86-
return Status::Invalid("Should call Commit first before GetLastCommitTableRequest.");
101+
return Status::Invalid(
102+
"Should call Commit first before GetLastCommitTableRequest. An attempt refused "
103+
"before it built a request leaves none behind.");
87104
}
88105
}
89106

0 commit comments

Comments
 (0)