Skip to content

Commit bd48508

Browse files
committed
feat: add ManageSnapshots.fast_forward_branch
test: fast_forward_branch auto-creates missing from_branch test: fast_forward_branch is a no-op when snapshots already equal test: fast_forward_branch rejects a tag as the source ref test: fast_forward_branch rejects missing to_ref test: fast_forward_branch rejects non-ancestor target test: fast_forward_branch preserves retention fields test: fast_forward_branch composes in a manage_snapshots chain test(integration): fast_forward_branch on real catalogs docs: add fast_forward_branch section with WAP example chore: apply ruff-format and add None-checks for mypy
1 parent 2c75523 commit bd48508

6 files changed

Lines changed: 392 additions & 0 deletions

File tree

mkdocs/docs/api.md

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1483,6 +1483,72 @@ Remove an existing branch:
14831483
table.manage_snapshots().remove_branch("dev").commit()
14841484
```
14851485

1486+
#### Fast-forwarding a branch
1487+
1488+
Fast-forward `from_branch` to point at the snapshot referenced by
1489+
`to_ref`. `to_ref` may be a branch or tag; `from_branch` must be a
1490+
branch. If `from_branch` does not yet exist it is created pointing at
1491+
`to_ref`'s snapshot. If both already point at the same snapshot the
1492+
call is a no-op. Otherwise `from_branch`'s current snapshot must be an
1493+
ancestor of `to_ref`'s snapshot; if not, `NotAncestorError` is raised.
1494+
1495+
```python
1496+
with table.manage_snapshots() as ms:
1497+
ms.fast_forward_branch("main", "audit-branch")
1498+
```
1499+
1500+
##### End-to-end: write-audit-publish
1501+
1502+
The canonical use case for fast-forward is the write-audit-publish
1503+
(WAP) pattern: writes proceed on a side branch, validation runs
1504+
against that branch, and only after validation succeeds is the main
1505+
branch advanced to publish the new data.
1506+
1507+
```python
1508+
import pyarrow as pa
1509+
import pyarrow.compute as pc
1510+
from pyiceberg.catalog import load_catalog
1511+
1512+
catalog = load_catalog("prod")
1513+
table = catalog.load_table("sales.orders")
1514+
1515+
# 1. WRITE — create a side branch off main and append to it.
1516+
main_snapshot_id = table.current_snapshot().snapshot_id
1517+
table.manage_snapshots().create_branch(
1518+
snapshot_id=main_snapshot_id,
1519+
branch_name="audit",
1520+
).commit()
1521+
1522+
new_rows = pa.table({
1523+
"order_id": [1001, 1002, 1003],
1524+
"amount": [ 49.99, 129.00, 12.50],
1525+
})
1526+
table.append(new_rows, branch="audit")
1527+
1528+
# 2. AUDIT — scan the audit branch and run whatever validation
1529+
# your data-quality contract requires. Nothing on `main` has
1530+
# changed yet, so readers of `main` still see the pre-write state.
1531+
audit_snapshot_id = table.refs()["audit"].snapshot_id
1532+
audit_data = table.scan(snapshot_id=audit_snapshot_id).to_arrow()
1533+
1534+
assert audit_data.num_rows > 0, "audit branch is empty"
1535+
assert pc.all(pc.greater(audit_data["amount"], 0)).as_py(), \
1536+
"found non-positive amounts"
1537+
1538+
# 3. PUBLISH — validation passed; fast-forward main to audit.
1539+
# Because the audit branch was created from main and only appended
1540+
# to, main's current snapshot is an ancestor of audit's snapshot,
1541+
# so the fast-forward is valid.
1542+
with table.manage_snapshots() as ms:
1543+
ms.fast_forward_branch("main", "audit")
1544+
ms.remove_branch("audit") # optional: clean up the side branch
1545+
```
1546+
1547+
If validation fails, callers simply skip the fast-forward step. The
1548+
audit branch (and its data files) can then be inspected, rewritten,
1549+
or removed via `remove_branch` and subsequent snapshot expiration —
1550+
without ever having polluted `main`.
1551+
14861552
## Table Maintenance
14871553

14881554
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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,3 +138,15 @@ class WaitingForLockException(Exception):
138138

139139
class ValidationException(Exception):
140140
"""Raised when validation fails."""
141+
142+
143+
class NoSuchSnapshotRefError(ValueError):
144+
"""Raised when a named snapshot ref (branch or tag) does not exist."""
145+
146+
147+
class SnapshotRefTypeError(ValueError):
148+
"""Raised when an operation expects a branch and gets a tag (or vice versa)."""
149+
150+
151+
class NotAncestorError(ValueError):
152+
"""Raised when an operation requires ancestry between two snapshots and it does not hold."""

pyiceberg/table/update/snapshot.py

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,11 @@
2626
from typing import TYPE_CHECKING, Generic
2727

2828
from pyiceberg.avro.codecs import AvroCompressionCodec
29+
from pyiceberg.exceptions import (
30+
NoSuchSnapshotRefError,
31+
NotAncestorError,
32+
SnapshotRefTypeError,
33+
)
2934
from pyiceberg.expressions import AlwaysFalse, BooleanExpression, Or
3035
from pyiceberg.expressions.visitors import (
3136
ROWS_MIGHT_NOT_MATCH,
@@ -56,6 +61,7 @@
5661
SnapshotSummaryCollector,
5762
Summary,
5863
ancestors_of,
64+
is_ancestor_of,
5965
latest_ancestor_before_timestamp,
6066
update_snapshot_summaries,
6167
)
@@ -1028,6 +1034,70 @@ def _current_ancestors(self) -> set[int]:
10281034
)
10291035
}
10301036

1037+
def fast_forward_branch(self, from_branch: str, to_ref: str) -> ManageSnapshots:
1038+
"""Fast-forward ``from_branch`` to the snapshot referenced by ``to_ref``.
1039+
1040+
If ``from_branch`` does not exist, it is created pointing at ``to_ref``'s snapshot (Java/Spark parity).
1041+
If both refs already point to the same snapshot the call is a no-op.
1042+
Otherwise ``from_branch`` must be a branch (not a tag) and its current
1043+
snapshot must be an ancestor of ``to_ref``'s snapshot.
1044+
1045+
Note:
1046+
Unlike Java's ``ManageSnapshots.fastForwardBranch``, this method does not
1047+
provide Java-parity for intra-chain semantics.
1048+
1049+
1) Java maintains a mutable ``updatedRefs`` map that reflects prior operations
1050+
in the same chain, so calling ``createBranch`` and then ``fastForwardBranch``
1051+
on the same ref in one chain observes the freshly-created state.
1052+
1053+
2) pyiceberg's ``ManageSnapshots`` accumulates ``SetSnapshotRefUpdate`` values without
1054+
mutating ``table_metadata.refs`` between chained calls;
1055+
-> The no-op and ancestry checks below operate on committed metadata only.
1056+
-> Callers that need Java-parity behavior should commit between chain steps.
1057+
1058+
Args:
1059+
from_branch: name of the branch to advance.
1060+
to_ref: name of the branch or tag whose snapshot ``from_branch`` will point to.
1061+
1062+
Returns:
1063+
This for method chaining.
1064+
1065+
Raises:
1066+
NoSuchSnapshotRefError: ``to_ref`` does not exist.
1067+
SnapshotRefTypeError: ``from_branch`` exists but is a tag.
1068+
NotAncestorError: ``from_branch``'s snapshot is not an ancestor of ``to_ref``'s snapshot.
1069+
"""
1070+
refs = self._transaction.table_metadata.refs
1071+
1072+
if to_ref not in refs:
1073+
raise NoSuchSnapshotRefError(f"Ref does not exist: {to_ref}")
1074+
to_snapshot_id = refs[to_ref].snapshot_id
1075+
1076+
if from_branch not in refs:
1077+
return self.create_branch(snapshot_id=to_snapshot_id, branch_name=from_branch)
1078+
1079+
from_ref = refs[from_branch]
1080+
if from_ref.snapshot_ref_type != SnapshotRefType.BRANCH:
1081+
raise SnapshotRefTypeError(f"Ref {from_branch} is a tag, not a branch")
1082+
1083+
if from_ref.snapshot_id == to_snapshot_id:
1084+
return self
1085+
1086+
if not is_ancestor_of(to_snapshot_id, from_ref.snapshot_id, self._transaction.table_metadata):
1087+
raise NotAncestorError(f"Cannot fast-forward: {from_branch} is not an ancestor of {to_ref}")
1088+
1089+
update, requirement = self._transaction._set_ref_snapshot(
1090+
snapshot_id=to_snapshot_id,
1091+
ref_name=from_branch,
1092+
type=SnapshotRefType.BRANCH,
1093+
max_ref_age_ms=from_ref.max_ref_age_ms,
1094+
max_snapshot_age_ms=from_ref.max_snapshot_age_ms,
1095+
min_snapshots_to_keep=from_ref.min_snapshots_to_keep,
1096+
)
1097+
self._updates += update
1098+
self._requirements += requirement
1099+
return self
1100+
10311101

10321102
class ExpireSnapshots(UpdateTableMetadata["ExpireSnapshots"]):
10331103
"""Expire snapshots by ID.

tests/integration/test_snapshot_operations.py

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -282,6 +282,39 @@ def test_rollback_to_timestamp_no_valid_snapshot(table_with_snapshots: Table) ->
282282
table_with_snapshots.manage_snapshots().rollback_to_timestamp(timestamp_ms=oldest_timestamp).commit()
283283

284284

285+
@pytest.mark.integration
286+
@pytest.mark.parametrize("catalog", [lf("session_catalog_hive"), lf("session_catalog")])
287+
def test_fast_forward_branch(catalog: Catalog) -> None:
288+
identifier = "default.test_table_snapshot_operations"
289+
tbl = catalog.load_table(identifier)
290+
assert len(tbl.history()) > 2
291+
292+
# Create a side branch off an older snapshot on main, then append to it
293+
# so that the side branch is a strict descendant of main's older snapshot
294+
# but the *current* main is not yet caught up.
295+
current_snapshot = tbl.current_snapshot()
296+
assert current_snapshot is not None
297+
main_snapshot_id = current_snapshot.snapshot_id
298+
side_branch = "audit_ff"
299+
300+
tbl.manage_snapshots().create_branch(snapshot_id=main_snapshot_id, branch_name=side_branch).commit()
301+
302+
arrow_schema = tbl.schema().as_arrow()
303+
new_rows = pa.Table.from_pylist([{col.name: None for col in arrow_schema}], schema=arrow_schema)
304+
tbl.append(new_rows, branch=side_branch)
305+
306+
# Validate appending to the side branch has advanced its snapshot.
307+
tbl = catalog.load_table(identifier)
308+
audit_snapshot_id = tbl.refs()[side_branch].snapshot_id
309+
assert audit_snapshot_id != main_snapshot_id, "side-branch append should have advanced it"
310+
311+
# Fast-forward main to the side branch.
312+
tbl.manage_snapshots().fast_forward_branch(from_branch="main", to_ref=side_branch).commit()
313+
314+
tbl = catalog.load_table(identifier)
315+
assert tbl.refs()["main"].snapshot_id == audit_snapshot_id
316+
317+
285318
@pytest.mark.integration
286319
def test_rollback_to_timestamp(table_with_snapshots: Table) -> None:
287320
current_snapshot = table_with_snapshots.current_snapshot()

0 commit comments

Comments
 (0)