[python] Support delete_snapshot/snapshot_exists in SnapshotManager

docs/content/pypaimon/python-api.md

@@ -393,6 +393,13 @@ snapshot_manager = SnapshotManager(table)
 t1 = snapshot_manager.get_snapshot_by_id(1).time_millis
 t2 = snapshot_manager.get_snapshot_by_id(2).time_millis
 
+# You can also check if a snapshot exists
+if snapshot_manager.snapshot_exists(1):
+    print("Snapshot 1 exists")
+
+# Delete a specific snapshot
+snapshot_manager.delete_snapshot(3)
+
 # Read records committed between [t1, t2]
 table_inc = table.copy({CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP: f"{t1},{t2}"})
 

paimon-python/pypaimon/snapshot/snapshot_manager.py

@@ -134,6 +134,35 @@ def get_snapshot_path(self, snapshot_id: int) -> str:
         """Get the file path for the given snapshot ID."""
         return f"{self.snapshot_dir}/snapshot-{snapshot_id}"
 
+    def snapshot_exists(self, snapshot_id: int) -> bool:
+        """
+        Check if a snapshot with the given ID exists.
+
+        Args:
+            snapshot_id: The snapshot ID to check
+
+        Returns:
+            True if the snapshot exists, False otherwise
+        """
+        snapshot_file = self.get_snapshot_path(snapshot_id)
+        return self.file_io.exists(snapshot_file)
+
+    def delete_snapshot(self, snapshot_id: int) -> None:
+        """
+        Delete the snapshot file for the given ID.
+
+        Args:
+            snapshot_id: The snapshot ID to delete
+
+        Raises:
+            FileNotFoundError: If the snapshot file does not exist
+            IOError: If an error occurs during deletion
+        """
+        snapshot_file = self.get_snapshot_path(snapshot_id)
+        if not self.file_io.exists(snapshot_file):
+            raise FileNotFoundError(f"Snapshot file not found: {snapshot_file}")
+        self.file_io.delete(snapshot_file)
+
     def try_get_earliest_snapshot(self) -> Optional[Snapshot]:
         earliest_file = f"{self.snapshot_dir}/EARLIEST"
         if self.file_io.exists(earliest_file):

paimon-python/pypaimon/tests/snapshot_manager_test.py

@@ -138,6 +138,85 @@ def should_scan(snapshot):
         self.assertEqual(next_id, 8)  # 5 + 3 = 8, continue from here
         self.assertEqual(skipped_count, 3)  # All 3 were skipped
 
+    def test_snapshot_exists_returns_true_when_snapshot_exists(self):
+        """snapshot_exists should return True when snapshot file exists."""
+        from pypaimon.snapshot.snapshot_manager import SnapshotManager
+
+        table = Mock()
+        table.table_path = "/tmp/test_table"
+        table.file_io = Mock()
+        table.file_io.exists.return_value = True
+        table.catalog_environment = Mock()
+        table.catalog_environment.snapshot_loader.return_value = None
+
+        manager = SnapshotManager(table)
+
+        # Test existing snapshot
+        exists = manager.snapshot_exists(123)
+
+        self.assertTrue(exists)
+        # Verify the correct path was checked
+        table.file_io.exists.assert_called_once_with("/tmp/test_table/snapshot/snapshot-123")
+
+    def test_snapshot_exists_returns_false_when_snapshot_not_exists(self):
+        """snapshot_exists should return False when snapshot file does not exist."""
+        from pypaimon.snapshot.snapshot_manager import SnapshotManager
+
+        table = Mock()
+        table.table_path = "/tmp/test_table"
+        table.file_io = Mock()
+        table.file_io.exists.return_value = False
+        table.catalog_environment = Mock()
+        table.catalog_environment.snapshot_loader.return_value = None
+
+        manager = SnapshotManager(table)
+
+        # Test non-existing snapshot
+        exists = manager.snapshot_exists(999)
+
+        self.assertFalse(exists)
+        table.file_io.exists.assert_called_once_with("/tmp/test_table/snapshot/snapshot-999")
+
+    def test_delete_snapshot_removes_snapshot_file(self):
+        """delete_snapshot should successfully delete existing snapshot file."""
+        from pypaimon.snapshot.snapshot_manager import SnapshotManager
+
+        table = Mock()
+        table.table_path = "/tmp/test_table"
+        table.file_io = Mock()
+        table.file_io.exists.return_value = True
+        table.catalog_environment = Mock()
+        table.catalog_environment.snapshot_loader.return_value = None
+
+        manager = SnapshotManager(table)
+
+        # Test deleting existing snapshot
+        manager.delete_snapshot(456)
+
+        # Verify delete was called with correct path
+        table.file_io.delete.assert_called_once_with("/tmp/test_table/snapshot/snapshot-456")
+
+    def test_delete_snapshot_raises_error_when_snapshot_not_exists(self):
+        """delete_snapshot should raise FileNotFoundError when snapshot does not exist."""
+        from pypaimon.snapshot.snapshot_manager import SnapshotManager
+
+        table = Mock()
+        table.table_path = "/tmp/test_table"
+        table.file_io = Mock()
+        table.file_io.exists.return_value = False
+        table.catalog_environment = Mock()
+        table.catalog_environment.snapshot_loader.return_value = None
+
+        manager = SnapshotManager(table)
+
+        # Test deleting non-existing snapshot should raise error
+        with self.assertRaises(FileNotFoundError) as context:
+            manager.delete_snapshot(789)
+
+        self.assertIn("Snapshot file not found", str(context.exception))
+        # Verify delete was not called
+        table.file_io.delete.assert_not_called()
+
 
 if __name__ == '__main__':
     unittest.main()