refactor: PingingPool pings sessions using SELECT 1 (#75)

Currently, PingingPool pings sessions in the background by calling `session.exists()` which calls `GetSession`. 

Using `SELECT 1` is preferred and is used in other client libraries such as [Go](https://github.com/googleapis/google-cloud-go/blob/53898305c6f21b3c3eef34fcff6c61a2cb36f602/spanner/session.go#L227):

Author: larkee

google/cloud/spanner_v1/pool.py

@@ -314,7 +314,7 @@ class PingingPool(AbstractSessionPool):
     - Sessions are used in "round-robin" order (LRU first).
 
     - "Pings" existing sessions in the background after a specified interval
-      via an API call (``session.exists()``).
+      via an API call (``session.ping()``).
 
     - Blocks, with a timeout, when :meth:`get` is called on an empty pool.
       Raises after timing out.
@@ -387,6 +387,9 @@ def get(self, timeout=None):  # pylint: disable=arguments-differ
         ping_after, session = self._sessions.get(block=True, timeout=timeout)
 
         if _NOW() > ping_after:
+            # Using session.exists() guarantees the returned session exists.
+            # session.ping() uses a cached result in the backend which could
+            # result in a recently deleted session being returned.
             if not session.exists():
                 session = self._new_session()
                 session.create()
@@ -430,7 +433,9 @@ def ping(self):
                 # Re-add to queue with existing expiration
                 self._sessions.put((ping_after, session))
                 break
-            if not session.exists():  # stale
+            try:
+                session.ping()
+            except NotFound:
                 session = self._new_session()
                 session.create()
             # Re-add to queue with new expiration

google/cloud/spanner_v1/session.py

@@ -153,6 +153,17 @@ def delete(self):
 
         api.delete_session(self.name, metadata=metadata)
 
+    def ping(self):
+        """Ping the session to keep it alive by executing "SELECT 1".
+
+        :raises: ValueError: if :attr:`session_id` is not already set.
+        """
+        if self._session_id is None:
+            raise ValueError("Session ID not set by back-end")
+        api = self._database.spanner_api
+        metadata = _metadata_with_prefix(self._database.name)
+        api.execute_sql(self.name, "SELECT 1", metadata=metadata)
+
     def snapshot(self, **kw):
         """Create a snapshot to perform a set of reads with shared staleness.
 

tests/unit/test_pool.py

@@ -567,7 +567,7 @@ def test_ping_oldest_fresh(self):
 
         pool.ping()
 
-        self.assertFalse(SESSIONS[0]._exists_checked)
+        self.assertFalse(SESSIONS[0]._pinged)
 
     def test_ping_oldest_stale_but_exists(self):
         import datetime
@@ -584,7 +584,7 @@ def test_ping_oldest_stale_but_exists(self):
         with _Monkey(MUT, _NOW=lambda: later):
             pool.ping()
 
-        self.assertTrue(SESSIONS[0]._exists_checked)
+        self.assertTrue(SESSIONS[0]._pinged)
 
     def test_ping_oldest_stale_and_not_exists(self):
         import datetime
@@ -602,7 +602,7 @@ def test_ping_oldest_stale_and_not_exists(self):
         with _Monkey(MUT, _NOW=lambda: later):
             pool.ping()
 
-        self.assertTrue(SESSIONS[0]._exists_checked)
+        self.assertTrue(SESSIONS[0]._pinged)
         SESSIONS[1].create.assert_called()
 
 
@@ -850,6 +850,7 @@ def __init__(self, database, exists=True, transaction=None):
         self._database = database
         self._exists = exists
         self._exists_checked = False
+        self._pinged = False
         self.create = mock.Mock()
         self._deleted = False
         self._transaction = transaction
@@ -861,6 +862,13 @@ def exists(self):
         self._exists_checked = True
         return self._exists
 
+    def ping(self):
+        from google.cloud.exceptions import NotFound
+
+        self._pinged = True
+        if not self._exists:
+            raise NotFound("expired session")
+
     def delete(self):
         from google.cloud.exceptions import NotFound
 

tests/unit/test_session.py

@@ -210,6 +210,66 @@ def test_exists_error(self):
             metadata=[("google-cloud-resource-prefix", database.name)],
         )
 
+    def test_ping_wo_session_id(self):
+        database = self._make_database()
+        session = self._make_one(database)
+        with self.assertRaises(ValueError):
+            session.ping()
+
+    def test_ping_hit(self):
+        gax_api = self._make_spanner_api()
+        gax_api.execute_sql.return_value = "1"
+        database = self._make_database()
+        database.spanner_api = gax_api
+        session = self._make_one(database)
+        session._session_id = self.SESSION_ID
+
+        session.ping()
+
+        gax_api.execute_sql.assert_called_once_with(
+            self.SESSION_NAME,
+            "SELECT 1",
+            metadata=[("google-cloud-resource-prefix", database.name)],
+        )
+
+    def test_ping_miss(self):
+        from google.api_core.exceptions import NotFound
+
+        gax_api = self._make_spanner_api()
+        gax_api.execute_sql.side_effect = NotFound("testing")
+        database = self._make_database()
+        database.spanner_api = gax_api
+        session = self._make_one(database)
+        session._session_id = self.SESSION_ID
+
+        with self.assertRaises(NotFound):
+            session.ping()
+
+        gax_api.execute_sql.assert_called_once_with(
+            self.SESSION_NAME,
+            "SELECT 1",
+            metadata=[("google-cloud-resource-prefix", database.name)],
+        )
+
+    def test_ping_error(self):
+        from google.api_core.exceptions import Unknown
+
+        gax_api = self._make_spanner_api()
+        gax_api.execute_sql.side_effect = Unknown("testing")
+        database = self._make_database()
+        database.spanner_api = gax_api
+        session = self._make_one(database)
+        session._session_id = self.SESSION_ID
+
+        with self.assertRaises(Unknown):
+            session.ping()
+
+        gax_api.execute_sql.assert_called_once_with(
+            self.SESSION_NAME,
+            "SELECT 1",
+            metadata=[("google-cloud-resource-prefix", database.name)],
+        )
+
     def test_delete_wo_session_id(self):
         database = self._make_database()
         session = self._make_one(database)