feat(catalog): commit snapshots through a version-managed catalog

docs/source/user_guide/catalog.rst

@@ -113,7 +113,118 @@ Databases and tables are then created, listed, loaded, renamed and dropped
 through the regular ``Catalog`` API, and table snapshots can be listed through
 ``Catalog::ListSnapshots``.
 
-The C++ REST catalog covers the database, table and snapshot 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.
+Committing through the catalog
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+A REST catalog reports ``Catalog::SupportsVersionManagement() == true`` and
+publishes snapshots through the server. Pass the catalog and table identifier
+to ``CommitContextBuilder::WithCatalog`` to commit on local or object storage
+without an atomic snapshot-file rename:
+
+.. code-block:: cpp
+
+   // Writers and committers share ownership of the catalog.
+   std::shared_ptr<paimon::Catalog> shared_catalog(std::move(catalog));
+
+   // Load the table's identity before preparing changes.
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Table> table,
+                          shared_catalog->GetTable(paimon::Identifier("db", "tbl")));
+
+   paimon::CommitContextBuilder builder(table_path, "commit-user");
+   builder.WithCatalog(shared_catalog, paimon::Identifier("db", "tbl"));
+   // CatalogUuid() stays null when the server supplies no id.
+   if (std::optional<std::string> table_uuid = table->CatalogUuid()) {
+       builder.WithTableId(table_uuid.value());
+   }
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::CommitContext> context, builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FileStoreCommit> commit,
+                          paimon::FileStoreCommit::Create(std::move(context)));
+   PAIMON_RETURN_NOT_OK(commit->Commit(commit_messages));
+
+Configure native writers with the same catalog and identifier. They load the
+current schema and latest snapshot from the catalog, including committed
+offsets when a real-time writer is created:
+
+.. code-block:: cpp
+
+   paimon::WriteContextBuilder write_builder(table_path, "commit-user");
+   write_builder.WithCatalog(shared_catalog, paimon::Identifier("db", "tbl"));
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::WriteContext> write_context,
+                          write_builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FileStoreWrite> writer,
+                          paimon::FileStoreWrite::Create(std::move(write_context)));
+
+For real-time writes, also set ``WithStreamingMode(true)`` and
+``WithRealtimeContext(...)`` with ``realtime.enabled=true``. After
+``CommitWithProgress`` succeeds, pass its returned snapshot id to
+``writer->RefreshCommittedSnapshot`` to release committed in-memory data.
+The refresh reads the current snapshot from the catalog. If another commit has
+already advanced the latest snapshot, the requested snapshot must be published
+under the table path. Recreating the writer restores offsets from the catalog's
+current snapshot.
+
+Both builders use the catalog's file system for manifests and data, preserving
+its object-store credentials. ``WithFileSystem`` overrides it; writers also
+allow ``WithFileSystemSchemeToIdentifierMap`` to override file-system selection.
+For format tables, use ``WriteContextBuilder(FormatTable)`` instead.
+
+The snapshot being committed carries a uuid generated on the client, and the
+commit names the snapshot it is based on by that snapshot's uuid, so the server
+can tell a commit that raced another. A commit which lost such a race is
+rebased on the new latest snapshot and retried, bounded by ``commit.max-retries``
+and ``commit.timeout`` as a file-system commit is.
+
+The HTTP client neither retries nor follows redirects for a commit request.
+Transport errors and non-success responses, including 429, 503 and redirects,
+leave the outcome uncertain: the server may already have accepted the snapshot.
+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.
+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.
+
+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.
+
+.. note::
+
+   Manifests, data and historical metadata still require file-system access:
+
+   * ``FilterAndCommit``, ``CommitWithProgress`` and streaming writer recovery
+     may walk older snapshots to find the commit user's last snapshot. Those
+     snapshots must be published under the table path. A missing retained
+     snapshot is an error; the ``EARLIEST`` hint distinguishes expired history.
+   * ``RowIdCheckConflict`` reuses the latest snapshot from the catalog and reads
+     earlier snapshots from the table path. With data evolution enabled, the
+     check also requires the schemas recorded by the row-id files it inspects.
+     Publishing only the current schema is insufficient when files use older ids.
+   * ``RollbackToAsLatest`` reads its target snapshot from the table path.
+     ``Expire`` manages only snapshots published there and updates the
+     ``EARLIEST`` hint; it does not expire catalog-held metadata.
+     Before deleting files, expiration verifies that the catalog's current
+     snapshot matches the latest published snapshot and reads the retained
+     history. It skips expiration while the current snapshot is unpublished
+     or differs from its file-system copy. Catalog and retained-metadata read
+     errors stop expiration before deletion. A catalog that does not support
+     loading its current snapshot cannot perform local expiration.
+     Retained snapshots with index manifests also stop expiration with
+     ``NotImplemented`` before any files are deleted.
+
+Expiration preserves files referenced by retained snapshots, including files
+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.
+
+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.

docs/source/user_guide/format_table.rst

@@ -68,7 +68,7 @@ a constructor taking one - which is what Java does through ``FormatTable.newRead
 
 The table already carries its schema and the file system it was loaded through, so a context built
 from one refuses ``SetTableSchema()``, ``WithFileSystem()``,
-``WithFileSystemSchemeToIdentifierMap()`` and a branch rather than quietly ignoring them. Options
+``WithFileSystemSchemeToIdentifierMap()``, ``WithCatalog()`` and a branch. Options
 given at the call still win over the ones the schema stored, as they do everywhere else.
 
 A batch comes back in the table's column order, or the projection's when the read names one,
@@ -126,11 +126,13 @@ A setting the format path cannot act on is refused by name rather than quietly d
 * ``WriteContextBuilder::WithWriteSchema()``, which names a subset of the columns to write.
 * ``WriteContextBuilder::WithWriteId()``, which prefixes a postpone-bucket writer's files so one
   compaction reader can put them back in order; a format table has no buckets.
-* ``CommitContextBuilder::IgnoreEmptyCommit(false)``, ``UseRESTCatalogCommit(true)`` and
-  ``AppendCommitCheckConflict(true)``. Keeping an empty commit means writing a snapshot that adds
-  no files, a rest-catalog commit sends that snapshot to a catalog, and the conflict check reads
-  the manifests of concurrent commits - none of which exist here. Each is refused only when set
-  away from its default, so an ordinary commit is unaffected.
+* ``CommitContextBuilder::IgnoreEmptyCommit(false)``, ``UseRESTCatalogCommit(true)``,
+  ``WithTableId()``, ``WithCatalog()`` and ``AppendCommitCheckConflict(true)``. Keeping an empty
+  commit means writing a snapshot that adds no files, a rest-catalog commit sends that snapshot to
+  a catalog, a table id names the table in the request that commit sends, committing through a
+  catalog hands it a snapshot to take, and the conflict check reads the manifests of concurrent
+  commits - none of which exist here. Each is refused only when set away from its default, so an
+  ordinary commit is unaffected.
 * the ``branch`` option, whatever names it: a format table keeps no metadata to branch, and its
   data is the files under its one location, which is where a read or a write would go whichever
   branch was asked for. It is refused wherever it comes from - the schema, a catalog that parsed

docs/source/user_guide/snapshot.rst

@@ -49,19 +49,20 @@ Snapshot File
 Snapshot file is JSON and includes:
 
 1. ``version``: Snapshot file version, current is 3.
-2. ``id``: Snapshot id, same as the file name.
-3. ``schemaId``: The corresponding schema version for this commit.
-4. ``baseManifestList``: A manifest list recording all changes from the previous snapshots.
-5. ``deltaManifestList``: A manifest list recording all new changes occurred in this snapshot.
-6. ``changelogManifestList``: A manifest list recording all changelog produced in this snapshot; ``null`` if no changelog is produced.
-7. ``indexManifest``: A manifest recording all index files of this table; ``null`` if no table index file exists.
-8. ``commitUser``: Usually generated by UUID; used for recovery of streaming writes—one stream write job with one user.
-9. ``commitIdentifier``: Transaction id corresponding to streaming write; each transaction may result in multiple commits for different ``commitKind`` values.
-10. ``commitKind``: Type of changes in this snapshot, including ``append``, ``compact``, ``overwrite`` and ``analyze``.
-11. ``timeMillis``: Commit time in milliseconds.
-12. ``logOffsets``: Commit log offsets.
-13. ``totalRecordCount``: Record count of all changes occurred in this snapshot.
-14. ``deltaRecordCount``: Record count of all new changes occurred in this snapshot.
-15. ``changelogRecordCount``: Record count of all changelog produced in this snapshot.
-16. ``watermark``: Watermark for input records, from Flink watermark mechanism; ``Long.MIN_VALUE`` if there is no watermark.
-17. ``statistics``: Stats file name for statistics of this table.
+2. ``uuid``: Uuid of this snapshot, generated on the client when the snapshot is created and absent only for a snapshot written before uuid was introduced. A commit names the uuid of the snapshot it is based on, so that a catalog can refuse a commit which raced another.
+3. ``id``: Snapshot id, same as the file name.
+4. ``schemaId``: The corresponding schema version for this commit.
+5. ``baseManifestList``: A manifest list recording all changes from the previous snapshots.
+6. ``deltaManifestList``: A manifest list recording all new changes occurred in this snapshot.
+7. ``changelogManifestList``: A manifest list recording all changelog produced in this snapshot; ``null`` if no changelog is produced.
+8. ``indexManifest``: A manifest recording all index files of this table; ``null`` if no table index file exists.
+9. ``commitUser``: Usually generated by UUID; used for recovery of streaming writes—one stream write job with one user.
+10. ``commitIdentifier``: Transaction id corresponding to streaming write; each transaction may result in multiple commits for different ``commitKind`` values.
+11. ``commitKind``: Type of changes in this snapshot, including ``append``, ``compact``, ``overwrite`` and ``analyze``.
+12. ``timeMillis``: Commit time in milliseconds.
+13. ``logOffsets``: Commit log offsets.
+14. ``totalRecordCount``: Record count of all changes occurred in this snapshot.
+15. ``deltaRecordCount``: Record count of all new changes occurred in this snapshot.
+16. ``changelogRecordCount``: Record count of all changelog produced in this snapshot.
+17. ``watermark``: Watermark for input records, from Flink watermark mechanism; ``Long.MIN_VALUE`` if there is no watermark.
+18. ``statistics``: Stats file name for statistics of this table.

docs/source/user_guide/write.rst

@@ -169,9 +169,17 @@ Operational Flow
 
 5. For a direct file-system commit on a non-object-store path, the engine
    passes the objects to ``FileStoreCommit`` for either an Append or PK table.
-   For an object-store path, it enables REST catalog commit mode, calls
-   ``Commit``, obtains the JSON request from ``GetLastCommitTableRequest``, and
-   sends that request to the REST catalog.
+   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.
 
 6. The local committer or REST catalog validates the messages, updates
    manifests/metadata, and finalizes the snapshot atomically.

include/paimon/catalog/catalog.h

@@ -44,8 +44,6 @@ class FormatTable;
 class Table;
 class View;
 class Schema;
-class Snapshot;
-class PartitionStatistics;
 class Tag;
 class Identifier;
 /// This interface is responsible for reading and writing metadata such as database/table from a
@@ -211,6 +209,11 @@ class PAIMON_EXPORT Catalog {
     virtual Result<std::vector<SnapshotInfo>> ListSnapshots(
         const Identifier& identifier, const std::string& branch = "") const = 0;
 
+    /// Returns whether this catalog publishes snapshots through catalog commits. Defaults to false.
+    /// @note Implementations returning true must also implement the internal
+    /// `VersionManagedCatalog`.
+    virtual bool SupportsVersionManagement() const;
+
     /// Gets a format table: a directory of data files laid out like a standard Hive table.
     ///
     /// A format table carries no snapshots and no manifests, so it is loaded through its own

include/paimon/catalog/table.h

@@ -20,6 +20,7 @@
 
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -45,9 +46,10 @@ class PAIMON_EXPORT Table {
                                                  const std::string& table_path,
                                                  const Identifier& identifier);
 
+    /// @param uuid Metastore UUID, or empty if unavailable.
     Table(const std::shared_ptr<Schema>& schema, const std::string& database,
-          const std::string& table_name)
-        : schema_(schema), database_(database), table_name_(table_name) {}
+          const std::string& table_name, const std::string& uuid = "")
+        : schema_(schema), database_(database), table_name_(table_name), uuid_(uuid) {}
 
     ~Table() = default;
 
@@ -59,10 +61,16 @@ class PAIMON_EXPORT Table {
     /// Full name of the table, default is database.tableName.
     std::string FullName() const;
 
-    /// UUID of the table, metastore can provide the true UUID of this table, default is the full
-    /// name.
+    /// Returns the metastore UUID, falling back to `FullName()`.
+    /// @note Use `CatalogUuid()` for commit requests to avoid sending a table name as an ID.
     std::string Uuid() const {
-        return FullName();
+        return uuid_.empty() ? FullName() : uuid_;
+    }
+
+    /// Returns the metastore UUID, or null if unavailable, for
+    /// `CommitContextBuilder::WithTableId()`.
+    std::optional<std::string> CatalogUuid() const {
+        return uuid_.empty() ? std::nullopt : std::optional<std::string>(uuid_);
     }
 
     /// Loads the latest schema of table.
@@ -74,6 +82,7 @@ class PAIMON_EXPORT Table {
     std::shared_ptr<Schema> schema_;
     std::string database_;
     std::string table_name_;
+    std::string uuid_;
 };
 
 }  // namespace paimon