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

[SteNicholas]

Purpose

Linked issue: close #326

Java Paimon commits a snapshot through Catalog#commitSnapshot(identifier, tableUuid, baseSnapshotUuid, snapshot, statistics), and CommitTableRequest carries tableId, baseSnapshotUuid, snapshot and statistics. Paimon C++ only had a request-only mode: CatalogSnapshotCommit built a request with snapshot and statistics, the caller had to send it, SnapshotCommit::Commit had no base snapshot uuid to pass down, and Snapshot had no uuid at all, so a catalog could neither detect a commit aimed at a recreated table nor perform the optimistic-concurrency check Java relies on.

This PR ports catalog-managed commits end to end, so a RestCatalog table is written, committed and expired with the same FileStoreWrite / FileStoreCommit API as a file-system table.

Catalog integration

• Catalog::SupportsVersionManagement() (default false) and the internal VersionManagedCatalog interface with LoadSnapshot() and CommitSnapshot(). A catalog that reports version management without implementing the interface is refused at commit creation instead of silently writing a snapshot file.
• RestCatalog implements it through RESTApi::LoadSnapshot() (GET .../tables/{table}/snapshot) and RESTApi::CommitSnapshot() (POST .../tables/{table}/commit), with the matching GetTableSnapshotResponse and CommitTableResponse messages. FileSystemCatalog does not manage versions, so it keeps the renaming file-system commit.
• CommitContextBuilder::WithCatalog(catalog, identifier) and WriteContextBuilder::WithCatalog(catalog, identifier). Commits and writers load the current schema from the catalog on every commit; when the catalog manages versions they also load its latest snapshot, through a SnapshotManager snapshot loader, and publish new snapshots through it. Only the main branch is supported, and the catalog supplies the file system for manifests and data unless WithFileSystem() overrides it. WithCatalog() and UseRESTCatalogCommit() are mutually exclusive.

Request and snapshot format, aligned with Java

• CommitTableRequest gains nullable tableId and baseSnapshotUuid in Java's field order, serialized as null when unset. SnapshotCommit::Commit(base_snapshot_uuid, snapshot, statistics) carries the base uuid; RenamingSnapshotCommit ignores it as Java does.
• Snapshot gains a nullable uuid, serialized right after version and omitted when unset. Snapshot::GenerateUuid() gives every new snapshot one, Tag::TrimToSnapshot() keeps it, and Table::CatalogUuid() exposes the metastore uuid for tableId.

Conflicts and recovery

• A catalog that refuses the commit (returns false) is a conflict: the commit rebases on the catalog's latest snapshot and retries within the existing commit.max-retries / commit.timeout limits, the same way a lost file-system rename does. ConflictDetection reuses the base snapshot of the attempt, since the latest snapshot may exist only in the catalog.
• A catalog error leaves the outcome uncertain: the manifests and snapshot metadata are kept and FilterAndCommit() resolves it, deduplicating against the catalog's snapshot and the published history. RestHttpClient::Execute() gains a retry_safe flag; the commit POST is sent with it off, so it is neither retried automatically nor replayed through a redirect.

Expiration

• When a catalog is configured, Expire() requires the catalog's current snapshot to be loaded and its retained history to be published on the file system before deleting anything. An unpublished or mismatched current snapshot skips expiration; catalog or retained-metadata read errors propagate without deleting files.
• Files referenced by retained snapshots are protected, including files a retained rollback snapshot adds back together with their extra files and external paths.
• The existing NotImplemented for index manifests is returned before any file is deleted, replacing the assert(false) that preceded it.

Compatibility

• The request-only UseRESTCatalogCommit() mode and the legacy CommitContext constructor are kept. GetLastCommitTableRequest() now always includes tableId and baseSnapshotUuid, null when unset; callers that spliced a tableId into that JSON themselves should switch to CommitContextBuilder::WithTableId().

Tests

New UT (76 cases), plus MockCatalog in src/paimon/testing/mock/mock_catalog.h and snapshot_test_helper.h:

• catalog_snapshot_commit_test.cpp (new target): commit through the catalog, a lost race is not an error, catalog failures propagate, a catalog without version management is refused, request-only build without a catalog.
• commit_table_request_test.cpp: TestNullTableIdAndBaseSnapshotUuid. rest_messages_test.cpp: GetTableSnapshotResponseRoundTrip, CommitTableResponseRoundTrip.
• snapshot_test.cpp: TestGenerateUuid, TestUuid. tag_test.cpp: TestTrimToSnapshotKeepsTheUuid.
• file_system_catalog_test.cpp: TestDoesNotManageVersions, TestClaimingVersionManagementWithoutImplementingItIsRefused.
• commit_context_test.cpp / write_context_test.cpp: TestWithCatalog, TestLegacyConstructor, TestCatalogRequiresMainBranch.
• file_store_commit_impl_test.cpp (37 cases): base snapshot uuid on the request; catalog commit takes the snapshot and is refused for a table recreated under the same name; catalog schema wins over the table path and is re-read on every commit; retries when refused, rebases on the winner of each race, succeeds after losing one race; unknown outcome keeps the manifests and recovers through FilterAndCommit, including an answer lost on the way back and a history that is only partly published; realtime commits are idempotent through the catalog; row-id checks use the catalog latest and the published history; rollback through the catalog; FileStoreRollbackExpireTest.TestRetainedRollbackKeepsRestoredFiles; catalog expiration waits for rollback publication, uses published snapshots and validates retained metadata before deleting files.
• snapshot_manager_test.cpp (11 cases): the snapshot loader answers which snapshot is latest, LatestSnapshotOfUser* walks the published history, stops at the earliest retained or an expired snapshot, retries a stale boundary and reports a gap it cannot explain.
• rest_catalog_test.cpp: CommitSnapshot, LoadSnapshot, CommitSnapshotErrors, FileStoreCommitIsBuiltFromTheCatalog, TableUuidFallsBackToFullNameWithoutServerId, CommitTakenThenReportedUnavailableIsNotReplayed, and RestCatalogCommitRecoveryTest.CommitAcceptedThenLostRecoversAfterRestart.
• rest_http_client_test.cpp: NonRetrySafeRequestDoesNotRetryTransientResponses, NonRetrySafeRequestDoesNotFollowRedirects.
• file_store_write_test.cpp: TestRealtimeCatalogCommitRefreshAndRecovery. expire_snapshots_test.cpp: TestCleanUnusedDataFileDeletesExtraFiles becomes TestCleanUnusedDataFilePreservesRetainedExtraFiles.

API and Format

Yes.

• include/paimon/catalog/catalog.h: Catalog::SupportsVersionManagement().
• include/paimon/catalog/table.h: optional uuid constructor argument and Table::CatalogUuid(); Table::Uuid() returns the catalog uuid when present and still falls back to the full name.
• include/paimon/commit_context.h, include/paimon/write_context.h: WithCatalog(catalog, identifier) on both builders, GetCatalog() / GetIdentifier() / GetTableId() accessors, and a CommitContext constructor taking the catalog fields; the previous constructor is retained.
• include/paimon/file_store_commit.h, include/paimon/file_store_write.h: documentation of the catalog commit mode, conflict retries, recovery and expiration requirements.
• Snapshot JSON gains an optional uuid field after version, omitted when unset. CommitTableRequest JSON gains tableId and baseSnapshotUuid, null when unset. The REST client now calls the /snapshot and /commit table endpoints of the Java REST catalog protocol.

Documentation

Yes.

• docs/source/user_guide/catalog.rst: a new Committing through the catalog section covering WithCatalog(), the schema and snapshot sources, conflict retries, recovery, the expiration requirements, the rollback / expiration coordination note, and the list of Java REST catalog operations that still have no C++ counterpart.
• docs/source/user_guide/snapshot.rst: the uuid field and the renumbered field list.
• docs/source/user_guide/write.rst: the catalog write and commit flow, and the request-only alternative for callers without a catalog client.
• docs/source/user_guide/format_table.rst: WithCatalog() and WithTableId() added to the builder options a format table refuses.

Generative AI tooling

Generated-by: Claude Code (claude-fable-5-1)

🤖 Generated with Claude Code

[lucasfang]

+1

[lxy-9602]

+1