Repository: pybel/pybel

Issue:
Improve efficiency of network drop
Right now, it's so slow that the Selventa large corpus can't be dropped in a reasonable time. 

Need to get smarter about building queries that efficiently find edges that are in a give network (to be dropped) but not in any other ones, then drop them.

Then, drop the network.

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/pybel/pybel/pull/319?src=pr&el=h1) Report
> :exclamation: No coverage uploaded for pull request base (`develop@45580a5`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference#section-missing-base-commit).
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/pybel/pybel/pull/319/graphs/tree.svg?width=650&src=pr&token=J7joRTRygG&height=150)](https://codecov.io/gh/pybel/pybel/pull/319?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             develop     #319   +/-   ##
==========================================
  Coverage           ?   87.66%           
==========================================
  Files              ?      124           
  Lines              ?     6122           
  Branches           ?      922           
==========================================
  Hits               ?     5367           
  Misses             ?      573           
  Partials           ?      182
```


| [Impacted Files](https://codecov.io/gh/pybel/pybel/pull/319?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [src/pybel/manager/cache\_manager.py](https://codecov.io/gh/pybel/pybel/pull/319/diff?src=pr&el=tree#diff-c3JjL3B5YmVsL21hbmFnZXIvY2FjaGVfbWFuYWdlci5weQ==) | `81.19% <100%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pybel/pybel/pull/319?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pybel/pybel/pull/319?src=pr&el=footer). Last update [45580a5...b2dc09f](https://codecov.io/gh/pybel/pybel/pull/319?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

scolby33: I modified the script to allow alternate connection strings and ran some timing tests on other RDBMS':

| Connection | Insert Time | Drop Time |
|----|----|----|
| `sqlite://` | 201.38 | 6.50 |
| `postgresql+psycopg2://scott@localhost/pybeltest` | 400.33 | 34.51 |

All times in seconds.

No statistics were done, but similar times were found for three runs with each DB.

With SQLite, I get a Python process at 99% CPU; with PostgreSQL, I get a Python process at ~45% and a Postgres process at about the same. Very interesting.

Patch:
diff --git a/src/pybel/manager/cache_manager.py b/src/pybel/manager/cache_manager.py
index c3f81dd5..b19921c7 100644
--- a/src/pybel/manager/cache_manager.py
+++ b/src/pybel/manager/cache_manager.py
@@ -27,7 +27,7 @@ from .exc import EdgeAddError
 from .lookup_manager import LookupManager
 from .models import (
     Annotation, AnnotationEntry, Author, Citation, Edge, Evidence, Modification, Namespace, NamespaceEntry, Network,
-    Node, Property,
+    Node, Property, network_edge, network_node, edge_property, edge_annotation
 )
 from .query_manager import QueryManager
 from .utils import extract_shared_optional, extract_shared_required, update_insert_values
@@ -637,28 +637,55 @@ class NetworkManager(NamespaceManager, AnnotationManager):
         """
         return self.session.query(exists().where(and_(Network.name == name, Network.version == version))).scalar()
 
-    @staticmethod
-    def iterate_singleton_edges_from_network(network):
-        """Gets all edges that only belong to the given network
+    def query_singleton_edges_from_network(self, network):
+        """Returns a query selecting all edge ids that only belong to the given network
 
         :type network: Network
-        :rtype: iter[Edge]
+        :rtype: sqlalchemy.orm.query.Query
         """
-        return (  # TODO implement with nested SQLAlchemy query for better speed
-            edge
-            for edge in network.edges
-            if edge.networks.count() == 1
+        ne1 = aliased(network_edge, name='ne1')
+        ne2 = aliased(network_edge, name='ne2')
+        singleton_edge_ids_for_network = (
+            self.session.query(ne1.c.edge_id)
+            .outerjoin(ne2, and_(
+                ne1.c.edge_id == ne2.c.edge_id,
+                ne1.c.network_id != ne2.c.network_id
+            ))
+            .filter(and_(
+                ne1.c.network_id == network.id,
+                ne2.c.edge_id == None
+            ))
         )
+        return singleton_edge_ids_for_network
 
     def drop_network(self, network):
         """Drops a network, while also cleaning up any edges that are no longer part of any network.
 
         :type network: Network
         """
-        for edge in self.iterate_singleton_edges_from_network(network):
-            self.session.delete(edge)
+        # get the IDs of the edges that will be orphaned by deleting this network
+        # FIXME: this list could be a problem if it becomes very large; possible optimization is a temporary table in DB
+        edge_ids = [result.edge_id for result in self.query_singleton_edges_from_network(network)]
+
+        # delete the network-to-node mappings for this network
+        self.session.query(network_node).filter(network_node.c.network_id == network.id).delete(synchronize_session=False)
+
+        # delete the edge-to-property mappings for the to-be-orphaned edges
+        self.session.query(edge_property).filter(edge_property.c.edge_id.in_(edge_ids)).delete(synchronize_session=False)
 
-        self.session.delete(network)
+        # delete the edge-to-annotation mappings for the to-be-orphaned edges
+        self.session.query(edge_annotation).filter(edge_annotation.c.edge_id.in_(edge_ids)).delete(synchronize_session=False)
+
+        # delete the edge-to-network mappings for this network
+        self.session.query(network_edge).filter(network_edge.c.network_id == network.id).delete(synchronize_session=False)
+
+        # delete the now-orphaned edges
+        self.session.query(Edge).filter(Edge.id.in_(edge_ids)).delete(synchronize_session=False)
+
+        # delete the network
+        self.session.query(Network).filter(Network.id == network.id).delete(synchronize_session=False)
+
+        # commit it!
         self.session.commit()
 
     def drop_network_by_id(self, network_id):
@@ -672,8 +699,7 @@ class NetworkManager(NamespaceManager, AnnotationManager):
     def drop_networks(self):
         """Drops all networks"""
         for network in self.session.query(Network).all():
-            self.session.delete(network)
-            self.session.commit()
+            self.drop_network(network)
 
     def get_network_versions(self, name):
         """Returns all of the versions of a network with the given name


Test patch:
diff --git a/scripts/run_drop_test.py b/scripts/run_drop_test.py
new file mode 100644
index 00000000..49584890
--- /dev/null
+++ b/scripts/run_drop_test.py
@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+
+"""Test loading and dropping a network."""
+
+import os
+import time
+from contextlib import contextmanager
+
+import click
+
+import pybel
+
+DEFAULT_CONNECTION = 'mysql+mysqldb://root@localhost/pbt?charset=utf8'
+PICKLE = 'small_corpus.bel.gpickle'
+SMALL_CORPUS_URL = 'https://arty.scai.fraunhofer.de/artifactory/bel/knowledge/selventa-small-corpus/selventa-small-corpus-20150611.bel'
+
+
+@contextmanager
+def time_me(start_string):
+    """Wrap statements with time logging."""
+    print(start_string)
+    parse_start_time = time.time()
+    yield
+    print(f'ran in {time.time() - parse_start_time:.2f} seconds')
+
+
+def get_numbers(graph, connection=None):
+    manager = pybel.Manager.from_connection(connection if connection else DEFAULT_CONNECTION)
+    print('inserting')
+    parse_start_time = time.time()
+    network = manager.insert_graph(graph)
+    print(f'inserted in {time.time() - parse_start_time:.2f} seconds')
+
+    print('dropping')
+    drop_start_time = time.time()
+    manager.drop_network(network)
+    drop_time = time.time() - drop_start_time
+    print(f'dropped in {drop_time:.2f} seconds')
+
+    return drop_time
+
+
+@click.command()
+@click.option('--connection', help=f'SQLAlchemy connection. Defaults to {DEFAULT_CONNECTION}')
+def main(connection):
+    """Parse a network, load it to the database, then test how fast it drops."""
+
+    if os.path.exists(PICKLE):
+        print(f'opening from {PICKLE}')
+        graph = pybel.from_pickle(PICKLE)
+    else:
+        with time_me(f'opening from {SMALL_CORPUS_URL}'):
+            manager = pybel.Manager.from_connection(connection if connection else DEFAULT_CONNECTION)
+            graph = pybel.from_url(SMALL_CORPUS_URL, manager=manager, use_tqdm=True, citation_clearing=False)
+
+        pybel.to_pickle(graph, PICKLE)
+
+    n = 1
+    # FIXME this fails if you do it with the same manager
+
+    times = [
+        get_numbers(graph, connection)
+        for _ in range(n)
+    ]
+
+    print(times)
+    print(sum(times) / n)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/test_manager/test_manager_drop.py b/tests/test_manager/test_manager_drop.py
index 35b09973..c7f65e37 100644
--- a/tests/test_manager/test_manager_drop.py
+++ b/tests/test_manager/test_manager_drop.py
@@ -90,10 +90,10 @@ class TestCascades(TemporaryCacheMixin):
         self.assertEqual(0, self.g2.edges.count())
 
     def test_get_orphan_edges(self):
-        edges = list(self.manager.iterate_singleton_edges_from_network(self.g1))
+        edges = [result.edge_id for result in self.manager.query_singleton_edges_from_network(self.g1)]
         self.assertEqual(2, len(edges))
-        self.assertIn(self.e2, edges)
-        self.assertIn(self.e3, edges)
+        self.assertIn(self.e2.id, edges)
+        self.assertIn(self.e3.id, edges)
 
     def test_drop_network_1(self):
         """When a network gets dropped, drop all of the edges if they don't appear in other networks"""
@@ -113,6 +113,13 @@ class TestCascades(TemporaryCacheMixin):
         self.assertEqual(1, self.manager.count_networks())
         self.assertEqual(3, self.g1.edges.count())
 
+    def test_drop_all_networks(self):
+        """When all networks are dropped, make sure all the edges and network_edge mappings are gone too"""
+        self.manager.drop_networks()
+
+        self.assertEqual(0, self.manager.count_edges())
+        self.assertEqual(0, self.manager.count_networks())
+
     def test_drop_modification(self):
         """Don't let this happen"""