@@ -1314,6 +1314,15 @@ def older_than(self, dt: datetime) -> ExpireSnapshots:
         return self
 
 
+def _active_files_count(manifests: list[ManifestFile]) -> int:
+    count = 0
+    for manifest in manifests:
+        if manifest.added_files_count is None or manifest.existing_files_count is None:
+            raise ValueError(f"Missing file counts in {manifest.manifest_path}")
+        count += manifest.added_files_count + manifest.existing_files_count
+    return count
+
+
 class RewriteManifests(_SnapshotProducer["RewriteManifests"]):
     """Rewrite the current snapshot's data manifests without changing data.
 
@@ -1370,6 +1379,9 @@ def rewrite_if(self, predicate: Callable[[ManifestFile], bool]) -> RewriteManife
         self._manifest_predicate = predicate
         return self
 
+    def _validate_concurrency(self) -> None:
+        """Skip validation; a retry replans against the refreshed branch head, so concurrent commits are kept."""
+
     def _deleted_entries(self) -> list[ManifestEntry]:
         return []
 
@@ -1419,6 +1431,7 @@ def _existing_manifests(self) -> list[ManifestFile]:
                 kept_manifests.append(manifest)
 
         new_manifests: list[ManifestFile] = []
+        rewritten_manifests: list[ManifestFile] = []
         for spec_id, manifests in data_manifests_by_spec.items():
             for group in self._group_by_target_size(manifests):
                 if len(group) == 1 and self._manifest_predicate is None:
@@ -1438,9 +1451,18 @@ def _existing_manifests(self) -> list[ManifestFile]:
                         self._entries_processed += 1
 
                 new_manifests.append(writer.to_manifest_file())
+                rewritten_manifests.extend(group)
                 self._created_count += 1
                 self._rewritten_count += len(group)
 
+        created_files_count = _active_files_count(new_manifests)
+        replaced_files_count = _active_files_count(rewritten_manifests)
+        if created_files_count != replaced_files_count:
+            raise ValidationException(
+                "Replaced and created manifests must have the same number of active files: "
+                f"{created_files_count} (new), {replaced_files_count} (old)"
+            )
+
         self._kept_count = len(kept_manifests)
         self.snapshot_properties = {
             **self.snapshot_properties,
 
@@ -15,13 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 from pathlib import Path
+from typing import Any
 
 import pyarrow as pa
 import pytest
 
 from pyiceberg.catalog import Catalog
 from pyiceberg.catalog.memory import InMemoryCatalog
-from pyiceberg.manifest import ManifestContent, ManifestFile
+from pyiceberg.exceptions import ValidationException
+from pyiceberg.manifest import ManifestContent, ManifestEntry, ManifestFile
 from pyiceberg.table import Table
 from pyiceberg.table.snapshots import Operation
 
@@ -303,3 +305,41 @@ def test_rewrites_needed_is_false_when_every_group_is_a_single_manifest(catalog:
     table = catalog.load_table("default.test_rewrite")
     assert table.current_snapshot() == snapshot_before
     assert len(_data_manifests(table)) == 3
+
+
+def test_rewrite_manifests_replans_after_concurrent_append(catalog: Catalog) -> None:
+    _create_table_with_appends(catalog, appends=3)
+    rewrite = catalog.load_table("default.test_rewrite").maintenance.rewrite_manifests()
+
+    # another writer commits between planning and committing the rewrite
+    catalog.load_table("default.test_rewrite").append(_arrow_table(offset=9))
+
+    rewrite.commit()
+
+    table = catalog.load_table("default.test_rewrite")
+    snapshot = table.current_snapshot()
+    assert snapshot is not None
+    assert snapshot.summary is not None
+    assert snapshot.summary.operation == Operation.REPLACE
+    # the concurrently appended file is merged in, not dropped
+    assert sorted(table.scan().to_arrow()["id"].to_pylist()) == list(range(1, 13))
+    assert len(_data_manifests(table)) == 1
+    assert snapshot.summary["manifests-replaced"] == "4"
+
+
+def test_rewrite_manifests_rejects_rewrite_that_drops_files(catalog: Catalog, monkeypatch: pytest.MonkeyPatch) -> None:
+    table = _create_table_with_appends(catalog, appends=3)
+    snapshot_before = table.current_snapshot()
+    lost_manifest_path = _data_manifests(table)[0].manifest_path
+    fetch_manifest_entry = ManifestFile.fetch_manifest_entry
+
+    def fetch_losing_one_manifest(self: ManifestFile, *args: Any, **kwargs: Any) -> list[ManifestEntry]:
+        entries = fetch_manifest_entry(self, *args, **kwargs)
+        return [] if self.manifest_path == lost_manifest_path else entries
+
+    monkeypatch.setattr(ManifestFile, "fetch_manifest_entry", fetch_losing_one_manifest)
+
+    with pytest.raises(ValidationException, match="same number of active files: 2 \\(new\\), 3 \\(old\\)"):
+        table.maintenance.rewrite_manifests().commit()
+
+    assert catalog.load_table("default.test_rewrite").current_snapshot() == snapshot_before