@@ -51,6 +51,7 @@
     ManifestEntry,
     ManifestEntryStatus,
     ManifestFile,
+    ManifestListWriterV3,
     ManifestWriter,
     write_manifest,
     write_manifest_list,
@@ -209,19 +210,6 @@ def delete_data_file(self, data_file: DataFile) -> _SnapshotProducer[U]:
         self._deleted_data_files.add(data_file)
         return self
 
-    def _calculate_added_rows(self, manifests: list[ManifestFile]) -> int:
-        """Calculate the number of added rows from a list of manifest files."""
-        added_rows = 0
-        for manifest in manifests:
-            if manifest.added_snapshot_id is None or manifest.added_snapshot_id == self._snapshot_id:
-                if manifest.added_rows_count is None:
-                    raise ValueError(
-                        "Cannot determine number of added rows in snapshot because "
-                        f"the entry for manifest {manifest.manifest_path} is missing the field `added-rows-count`"
-                    )
-                added_rows += manifest.added_rows_count
-        return added_rows
-
     @abstractmethod
     def _deleted_entries(self) -> list[ManifestEntry]: ...
 
@@ -335,6 +323,7 @@ def _commit(self) -> UpdatesAndRequirements:
         self._written_manifest_lists.append(manifest_list_file_path)
 
         first_row_id: int | None = None
+        added_rows: int | None = None
 
         if self._transaction.table_metadata.format_version >= 3:
             first_row_id = self._transaction.table_metadata.next_row_id
@@ -350,6 +339,10 @@ def _commit(self) -> UpdatesAndRequirements:
         ) as writer:
             writer.add_manifests(new_manifests)
 
+        if isinstance(writer, ManifestListWriterV3) and first_row_id is not None:
+            # row IDs assigned to manifests without a first_row_id, including existing rows of upgraded tables
+            added_rows = writer.next_row_id - first_row_id
+
         snapshot = Snapshot(
             snapshot_id=self._snapshot_id,
             parent_snapshot_id=self._parent_snapshot_id,
@@ -358,6 +351,7 @@ def _commit(self) -> UpdatesAndRequirements:
             summary=summary,
             schema_id=self._transaction.table_metadata.current_schema_id,
             first_row_id=first_row_id,
+            added_rows=added_rows,
         )
 
         add_snapshot_update = AddSnapshotUpdate(snapshot=snapshot)
 
@@ -2532,7 +2532,6 @@ def test_stage_only_overwrite_files(
     assert parent_snapshot_id == [None, first_snapshot, second_snapshot, second_snapshot, second_snapshot]
 
 
-@pytest.mark.skip("V3 writer support is not enabled.")
 @pytest.mark.integration
 def test_v3_write_and_read_row_lineage(spark: SparkSession, session_catalog: Catalog) -> None:
     """Test writing to a v3 table and reading with Spark."""
@@ -2569,6 +2568,14 @@ def test_v3_write_and_read_row_lineage(spark: SparkSession, session_catalog: Cat
         "Expected next_row_id to be incremented by the number of added rows"
     )
 
+    tbl.append(test_data)
+    assert tbl.metadata.next_row_id == initial_next_row_id + 2 * len(test_data)
+
+    rows = spark.sql(f"SELECT _row_id, _last_updated_sequence_number FROM {identifier} ORDER BY _row_id").collect()
+    assert [(row._row_id, row._last_updated_sequence_number) for row in rows] == [
+        (initial_next_row_id + i, 1 + i // len(test_data)) for i in range(2 * len(test_data))
+    ]
+
 
 # RecordBatchReader streaming append/overwrite — see https://github.com/apache/iceberg-python/issues/2152
 #
 
@@ -858,3 +858,72 @@ def test_manifest_list_has_no_first_row_id_for_v2(table_v2: Table) -> None:
 
     assert writer.call_args.kwargs["format_version"] == 2
     assert writer.call_args.kwargs["first_row_id"] is None
+
+
+@pytest.fixture
+def empty_table_v3(tmp_path: Path, arrow_table_simple: pa.Table) -> Table:
+    """An empty V3 table; writing V3 metadata files is not supported, so commits stay staged in a transaction."""
+    from pyiceberg.catalog.noop import NoopCatalog
+    from pyiceberg.io import load_file_io
+    from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC
+    from pyiceberg.table.metadata import new_table_metadata
+    from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
+
+    metadata = new_table_metadata(
+        schema=Catalog._convert_schema_if_needed(arrow_table_simple.schema, format_version=3),
+        partition_spec=UNPARTITIONED_PARTITION_SPEC,
+        sort_order=UNSORTED_SORT_ORDER,
+        location=str(tmp_path),
+        properties={"format-version": "3"},
+    ).model_copy(update={"next_row_id": 0})
+    return Table(
+        identifier=("default", "lineage"),
+        metadata=metadata,
+        metadata_location=f"{tmp_path}/metadata/00000.metadata.json",
+        io=load_file_io(),
+        catalog=NoopCatalog("NoopCatalog"),
+    )
+
+
+def test_v3_append_assigns_row_lineage(empty_table_v3: Table, arrow_table_simple: pa.Table) -> None:
+    txn = empty_table_v3.transaction()
+
+    txn.append(arrow_table_simple)
+    first = txn.table_metadata.current_snapshot()
+    assert first is not None
+    assert (first.first_row_id, first.added_rows) == (0, 3)
+    assert txn.table_metadata.next_row_id == 3
+
+    txn.append(arrow_table_simple)
+    second = txn.table_metadata.current_snapshot()
+    assert second is not None
+    assert (second.first_row_id, second.added_rows) == (3, 3)
+    assert txn.table_metadata.next_row_id == 6
+
+
+def test_v3_add_files_assigns_row_lineage(empty_table_v3: Table, arrow_table_simple: pa.Table, tmp_path: Path) -> None:
+    import pyarrow.parquet as pq
+
+    file_path = str(tmp_path / "data.parquet")
+    pq.write_table(arrow_table_simple, file_path)
+    txn = empty_table_v3.transaction()
+
+    txn.add_files([file_path])
+
+    snapshot = txn.table_metadata.current_snapshot()
+    assert snapshot is not None
+    assert (snapshot.first_row_id, snapshot.added_rows) == (0, 3)
+    assert txn.table_metadata.next_row_id == 3
+
+
+def test_v3_delete_assigns_row_ids_to_rewritten_rows_only(empty_table_v3: Table, arrow_table_simple: pa.Table) -> None:
+    txn = empty_table_v3.transaction()
+    txn.append(arrow_table_simple)
+
+    txn.delete("bar == 1")
+
+    snapshot = txn.table_metadata.current_snapshot()
+    assert snapshot is not None
+    # the rewritten data file holds the 2 surviving rows; the old file's DELETED entry uses no row IDs
+    assert (snapshot.first_row_id, snapshot.added_rows) == (3, 2)
+    assert txn.table_metadata.next_row_id == 5