Only send qid with PULL and DISCARD when necessary (#585)

Co-authored-by: Florent Biville <445792+fbiville@users.noreply.github.com>

Author: robsdedude

neo4j/io/__init__.py

@@ -137,6 +137,11 @@ class Bolt(abc.ABC):
     #: The pool of which this connection is a member
     pool = None
 
+    # Store the id of the most recent ran query to be able to reduce sent bits by
+    # using the default (-1) to refer to the most recent query when pulling
+    # results for it.
+    most_recent_qid = None
+
     def __init__(self, unresolved_address, sock, max_connection_lifetime, *, auth=None, user_agent=None, routing_context=None):
         self.unresolved_address = unresolved_address
         self.socket = sock

neo4j/work/result.py

@@ -48,11 +48,11 @@ def __init__(self, connection, on_error):
             connection raises of of the caught errors.
         :type on_error callable
         """
-        self._connection = connection
-        self._on_error = on_error
+        self.connection = connection
+        self.on_error = on_error
 
     def __getattr__(self, item):
-        connection_attr = getattr(self._connection, item)
+        connection_attr = getattr(self.connection, item)
         if not callable(connection_attr):
             return connection_attr
 
@@ -61,7 +61,7 @@ def inner(*args, **kwargs):
                 try:
                     func(*args, **kwargs)
                 except (Neo4jError, ServiceUnavailable, SessionExpired) as exc:
-                    self._on_error(exc)
+                    self.on_error(exc)
                     raise
             return inner
 
@@ -83,7 +83,7 @@ def __init__(self, connection, hydrant, fetch_size, on_closed,
         self._record_buffer = deque()
         self._summary = None
         self._bookmark = None
-        self._qid = -1
+        self._raw_qid = -1
         self._fetch_size = fetch_size
 
         # states
@@ -96,6 +96,13 @@ def __init__(self, connection, hydrant, fetch_size, on_closed,
         # the result has been fully iterated or consumed
         self._closed = False
 
+    @property
+    def _qid(self):
+        if self._raw_qid == self._connection.connection.most_recent_qid:
+            return -1
+        else:
+            return self._raw_qid
+
     def _tx_ready_run(self, query, parameters, **kwparameters):
         # BEGIN+RUN does not carry any extra on the RUN message.
         # BEGIN {extra}
@@ -117,7 +124,10 @@ def _run(self, query, parameters, db, access_mode, bookmarks, **kwparameters):
 
         def on_attached(metadata):
             self._metadata.update(metadata)
-            self._qid = metadata.get("qid", -1)  # For auto-commit there is no qid and Bolt 3 do not support qid
+            # For auto-commit there is no qid and Bolt 3 does not support qid
+            self._raw_qid = metadata.get("qid", -1)
+            if self._raw_qid != -1:
+                self._connection.connection.most_recent_qid = self._raw_qid
             self._keys = metadata.get("fields")
             self._attached = True
 

tests/unit/work/test_result.py

@@ -89,6 +89,7 @@ def __init__(self, records=None, run_meta=None, summary_meta=None,
             self._use_qid = force_qid
         self.fetch_idx = 0
         self._qid = -1
+        self.most_recent_qid = None
         self.record_idxs = [0] * len(self._records)
         self.to_pull = [None] * len(self._records)
         self._exhausted = [False] * len(self._records)