feat: add ManageSnapshots.fast_forward_branch

mkdocs/docs/api.md

@@ -1483,6 +1483,72 @@ Remove an existing branch:
 table.manage_snapshots().remove_branch("dev").commit()
 ```
 
+#### Fast-forwarding a branch
+
+Fast-forward `from_branch` to point at the snapshot referenced by
+`to_ref`. `to_ref` may be a branch or tag; `from_branch` must be a
+branch. If `from_branch` does not yet exist it is created pointing at
+`to_ref`'s snapshot. If both already point at the same snapshot the
+call is a no-op. Otherwise `from_branch`'s current snapshot must be an
+ancestor of `to_ref`'s snapshot; if not, `NotAncestorError` is raised.
+
+```python
+with table.manage_snapshots() as ms:
+    ms.fast_forward_branch("main", "audit-branch")
+```
+
+##### End-to-end: write-audit-publish
+
+The canonical use case for fast-forward is the write-audit-publish
+(WAP) pattern: writes proceed on a side branch, validation runs
+against that branch, and only after validation succeeds is the main
+branch advanced to publish the new data.
+
+```python
+import pyarrow as pa
+import pyarrow.compute as pc
+from pyiceberg.catalog import load_catalog
+
+catalog = load_catalog("prod")
+table = catalog.load_table("sales.orders")
+
+# 1. WRITE — create a side branch off main and append to it.
+main_snapshot_id = table.current_snapshot().snapshot_id
+table.manage_snapshots().create_branch(
+    snapshot_id=main_snapshot_id,
+    branch_name="audit",
+).commit()
+
+new_rows = pa.table({
+    "order_id": [1001, 1002, 1003],
+    "amount":   [ 49.99, 129.00, 12.50],
+})
+table.append(new_rows, branch="audit")
+
+# 2. AUDIT — scan the audit branch and run whatever validation
+#    your data-quality contract requires. Nothing on `main` has
+#    changed yet, so readers of `main` still see the pre-write state.
+audit_snapshot_id = table.refs()["audit"].snapshot_id
+audit_data = table.scan(snapshot_id=audit_snapshot_id).to_arrow()
+
+assert audit_data.num_rows > 0, "audit branch is empty"
+assert pc.all(pc.greater(audit_data["amount"], 0)).as_py(), \
+    "found non-positive amounts"
+
+# 3. PUBLISH — validation passed; fast-forward main to audit.
+#    Because the audit branch was created from main and only appended
+#    to, main's current snapshot is an ancestor of audit's snapshot,
+#    so the fast-forward is valid.
+with table.manage_snapshots() as ms:
+    ms.fast_forward_branch("main", "audit")
+    ms.remove_branch("audit")  # optional: clean up the side branch
+```
+
+If validation fails, callers simply skip the fast-forward step. The
+audit branch (and its data files) can then be inspected, rewritten,
+or removed via `remove_branch` and subsequent snapshot expiration —
+without ever having polluted `main`.
+
 ## Table Maintenance
 
 PyIceberg provides table maintenance operations through the `table.maintenance` API. This provides a clean interface for performing maintenance tasks like snapshot expiration.

pyiceberg/exceptions.py

@@ -138,3 +138,15 @@ class WaitingForLockException(Exception):
 
 class ValidationException(Exception):
     """Raised when validation fails."""
+
+
+class NoSuchSnapshotRefError(ValueError):
+    """Raised when a named snapshot ref (branch or tag) does not exist."""
+
+
+class SnapshotRefTypeError(ValueError):
+    """Raised when an operation expects a branch and gets a tag (or vice versa)."""
+
+
+class NotAncestorError(ValueError):
+    """Raised when an operation requires ancestry between two snapshots and it does not hold."""

pyiceberg/table/update/snapshot.py

@@ -26,6 +26,11 @@
 from typing import TYPE_CHECKING, Generic
 
 from pyiceberg.avro.codecs import AvroCompressionCodec
+from pyiceberg.exceptions import (
+    NoSuchSnapshotRefError,
+    NotAncestorError,
+    SnapshotRefTypeError,
+)
 from pyiceberg.expressions import AlwaysFalse, BooleanExpression, Or
 from pyiceberg.expressions.visitors import (
     ROWS_MIGHT_NOT_MATCH,
@@ -56,6 +61,7 @@
     SnapshotSummaryCollector,
     Summary,
     ancestors_of,
+    is_ancestor_of,
     latest_ancestor_before_timestamp,
     update_snapshot_summaries,
 )
@@ -1028,6 +1034,70 @@ def _current_ancestors(self) -> set[int]:
             )
         }
 
+    def fast_forward_branch(self, from_branch: str, to_ref: str) -> ManageSnapshots:
+        """Fast-forward ``from_branch`` to the snapshot referenced by ``to_ref``.
+
+        If ``from_branch`` does not exist, it is created pointing at ``to_ref``'s snapshot (Java/Spark parity).
+        If both refs already point to the same snapshot the call is a no-op.
+        Otherwise ``from_branch`` must be a branch (not a tag) and its current
+        snapshot must be an ancestor of ``to_ref``'s snapshot.
+
+        Note:
+            Unlike Java's ``ManageSnapshots.fastForwardBranch``, this method does not
+            provide Java-parity for intra-chain semantics.
+
+            1) Java maintains a mutable ``updatedRefs`` map that reflects prior operations
+            in the same chain, so calling ``createBranch`` and then ``fastForwardBranch``
+            on the same ref in one chain observes the freshly-created state.
+
+            2) pyiceberg's ``ManageSnapshots`` accumulates ``SetSnapshotRefUpdate`` values without
+            mutating ``table_metadata.refs`` between chained calls;
+            -> The no-op and ancestry checks below operate on committed metadata only.
+            -> Callers that need Java-parity behavior should commit between chain steps.
+
+        Args:
+            from_branch: name of the branch to advance.
+            to_ref: name of the branch or tag whose snapshot ``from_branch`` will point to.
+
+        Returns:
+            This for method chaining.
+
+        Raises:
+            NoSuchSnapshotRefError: ``to_ref`` does not exist.
+            SnapshotRefTypeError: ``from_branch`` exists but is a tag.
+            NotAncestorError: ``from_branch``'s snapshot is not an ancestor of ``to_ref``'s snapshot.
+        """
+        refs = self._transaction.table_metadata.refs
+
+        if to_ref not in refs:
+            raise NoSuchSnapshotRefError(f"Ref does not exist: {to_ref}")
+        to_snapshot_id = refs[to_ref].snapshot_id
+
+        if from_branch not in refs:
+            return self.create_branch(snapshot_id=to_snapshot_id, branch_name=from_branch)
+
+        from_ref = refs[from_branch]
+        if from_ref.snapshot_ref_type != SnapshotRefType.BRANCH:
+            raise SnapshotRefTypeError(f"Ref {from_branch} is a tag, not a branch")
+
+        if from_ref.snapshot_id == to_snapshot_id:
+            return self
+
+        if not is_ancestor_of(to_snapshot_id, from_ref.snapshot_id, self._transaction.table_metadata):
+            raise NotAncestorError(f"Cannot fast-forward: {from_branch} is not an ancestor of {to_ref}")
+
+        update, requirement = self._transaction._set_ref_snapshot(
+            snapshot_id=to_snapshot_id,
+            ref_name=from_branch,
+            type=SnapshotRefType.BRANCH,
+            max_ref_age_ms=from_ref.max_ref_age_ms,
+            max_snapshot_age_ms=from_ref.max_snapshot_age_ms,
+            min_snapshots_to_keep=from_ref.min_snapshots_to_keep,
+        )
+        self._updates += update
+        self._requirements += requirement
+        return self
+
 
 class ExpireSnapshots(UpdateTableMetadata["ExpireSnapshots"]):
     """Expire snapshots by ID.

tests/integration/test_snapshot_operations.py

@@ -282,6 +282,39 @@ def test_rollback_to_timestamp_no_valid_snapshot(table_with_snapshots: Table) ->
         table_with_snapshots.manage_snapshots().rollback_to_timestamp(timestamp_ms=oldest_timestamp).commit()
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [lf("session_catalog_hive"), lf("session_catalog")])
+def test_fast_forward_branch(catalog: Catalog) -> None:
+    identifier = "default.test_table_snapshot_operations"
+    tbl = catalog.load_table(identifier)
+    assert len(tbl.history()) > 2
+
+    # Create a side branch off an older snapshot on main, then append to it
+    # so that the side branch is a strict descendant of main's older snapshot
+    # but the *current* main is not yet caught up.
+    current_snapshot = tbl.current_snapshot()
+    assert current_snapshot is not None
+    main_snapshot_id = current_snapshot.snapshot_id
+    side_branch = "audit_ff"
+
+    tbl.manage_snapshots().create_branch(snapshot_id=main_snapshot_id, branch_name=side_branch).commit()
+
+    arrow_schema = tbl.schema().as_arrow()
+    new_rows = pa.Table.from_pylist([{col.name: None for col in arrow_schema}], schema=arrow_schema)
+    tbl.append(new_rows, branch=side_branch)
+
+    # Validate appending to the side branch has advanced its snapshot.
+    tbl = catalog.load_table(identifier)
+    audit_snapshot_id = tbl.refs()[side_branch].snapshot_id
+    assert audit_snapshot_id != main_snapshot_id, "side-branch append should have advanced it"
+
+    # Fast-forward main to the side branch.
+    tbl.manage_snapshots().fast_forward_branch(from_branch="main", to_ref=side_branch).commit()
+
+    tbl = catalog.load_table(identifier)
+    assert tbl.refs()["main"].snapshot_id == audit_snapshot_id
+
+
 @pytest.mark.integration
 def test_rollback_to_timestamp(table_with_snapshots: Table) -> None:
     current_snapshot = table_with_snapshots.current_snapshot()