Add support for v3 snapshot metadata fields

Author: geruh

pyiceberg/table/snapshots.py

@@ -22,7 +22,7 @@
 from enum import Enum
 from typing import TYPE_CHECKING, Any, DefaultDict, Dict, Iterable, List, Mapping, Optional
 
-from pydantic import Field, PrivateAttr, model_serializer
+from pydantic import Field, PrivateAttr, field_validator, model_serializer, model_validator
 
 from pyiceberg.io import FileIO
 from pyiceberg.manifest import DataFile, DataFileContent, ManifestFile, _manifests
@@ -237,14 +237,55 @@ def __eq__(self, other: Any) -> bool:
 
 
 class Snapshot(IcebergBaseModel):
+    """Represents a snapshot of an Iceberg table at a specific point in time.
+
+    A snapshot tracks the state of a table, including all data and delete files,
+    at the time the snapshot was created.
+    """
+
     snapshot_id: int = Field(alias="snapshot-id")
     parent_snapshot_id: Optional[int] = Field(alias="parent-snapshot-id", default=None)
     sequence_number: Optional[int] = Field(alias="sequence-number", default=INITIAL_SEQUENCE_NUMBER)
     timestamp_ms: int = Field(alias="timestamp-ms", default_factory=lambda: int(time.time() * 1000))
     manifest_list: str = Field(alias="manifest-list", description="Location of the snapshot's manifest list file")
+    first_row_id: Optional[int] = Field(
+        alias="first-row-id",
+        default=None,
+        description="The row-id of the first newly added row in this snapshot. Returns None when row lineage is not supported.",
+    )
+    added_rows: Optional[int] = Field(
+        alias="added-rows",
+        default=None,
+        description=(
+            "The upper bound of number of rows with assigned row IDs in this snapshot. Returns None if the value was not stored."
+        ),
+    )
     summary: Optional[Summary] = Field(default=None)
     schema_id: Optional[int] = Field(alias="schema-id", default=None)
 
+    @field_validator("first_row_id")
+    @classmethod
+    def validate_first_row_id(cls, v: Optional[int]) -> Optional[int]:
+        """Validate that first_row_id is non-negative if provided."""
+        if v is not None and v < 0:
+            raise ValueError(f"Invalid first-row-id (cannot be negative): {v}")
+        return v
+
+    @field_validator("added_rows")
+    @classmethod
+    def validate_added_rows(cls, v: Optional[int]) -> Optional[int]:
+        """Validate that added_rows is non-negative if provided."""
+        if v is not None and v < 0:
+            raise ValueError(f"Invalid added-rows (cannot be negative): {v}")
+        return v
+
+    @model_validator(mode="after")
+    def validate_row_lineage_fields(self) -> "Snapshot":
+        """Validate that added_rows is required when first_row_id is set."""
+        if self.first_row_id is not None and self.added_rows is None:
+            raise ValueError("Invalid added-rows (required when first-row-id is set): None")
+        return self
+
     def __str__(self) -> str:
         """Return the string representation of the Snapshot class."""
         operation = f"{self.summary.operation}: " if self.summary else ""

tests/table/test_snapshots.py

@@ -456,3 +456,78 @@ def test_ancestors_between(table_v2_with_extensive_snapshots: Table) -> None:
         )
         == 2000
     )
+
+
+def test_snapshot_v3_fields() -> None:
+    snapshot = Snapshot(
+        **{
+            "snapshot-id": 1,
+            "timestamp-ms": 1234567890,
+            "manifest-list": "s3:/a/b/c.avro",
+            "first-row-id": 100,
+            "added-rows": 1000,
+            "summary": {"operation": "append"},
+        }
+    )
+
+    assert snapshot.first_row_id == 100
+    assert snapshot.added_rows == 1000
+
+    actual = snapshot.model_dump_json()
+    expected = """{"snapshot-id":1,"sequence-number":0,"timestamp-ms":1234567890,"manifest-list":"s3:/a/b/c.avro","first-row-id":100,"added-rows":1000,"summary":{"operation":"append"}}"""
+    assert actual == expected
+
+
+def test_snapshot_v3_fields_validation_negative_first_row_id() -> None:
+    with pytest.raises(ValueError, match="Invalid first-row-id \\(cannot be negative\\): -1"):
+        Snapshot(
+            **{
+                "snapshot-id": 1,
+                "timestamp-ms": 1234567890,
+                "manifest-list": "s3:/a/b/c.avro",
+                "first-row-id": -1,
+                "added-rows": 1000,
+                "summary": {"operation": "append"},
+            }
+        )
+
+
+def test_snapshot_v3_fields_validation_negative_added_rows() -> None:
+    with pytest.raises(ValueError, match="Invalid added-rows \\(cannot be negative\\): -1"):
+        Snapshot(
+            **{
+                "snapshot-id": 1,
+                "timestamp-ms": 1234567890,
+                "manifest-list": "s3:/a/b/c.avro",
+                "first-row-id": 100,
+                "added-rows": -1,
+                "summary": {"operation": "append"},
+            }
+        )
+
+
+def test_snapshot_v3_fields_validation_first_row_id_requires_added_rows() -> None:
+    with pytest.raises(ValueError, match="Invalid added-rows \\(required when first-row-id is set\\): None"):
+        Snapshot(
+            **{
+                "snapshot-id": 1,
+                "timestamp-ms": 1234567890,
+                "manifest-list": "s3:/a/b/c.avro",
+                "first-row-id": 100,
+                "summary": {"operation": "append"},
+            }
+        )
+
+
+def test_snapshot_v3_fields_added_rows_without_first_row_id() -> None:
+    snapshot = Snapshot(
+        **{
+            "snapshot-id": 1,
+            "timestamp-ms": 1234567890,
+            "manifest-list": "s3:/a/b/c.avro",
+            "added-rows": 1000,
+            "summary": {"operation": "append"},
+        }
+    )
+    assert snapshot.first_row_id is None
+    assert snapshot.added_rows == 1000