Drop temporary tables when resetting a pooled connection

asyncpg/connection.py

@@ -1571,9 +1571,9 @@ async def reset(self, *, timeout=None):
         resembling that of a newly obtained connection.  Namely, an open
         transaction (if any) is rolled back, open cursors are closed,
         all `LISTEN <https://www.postgresql.org/docs/current/sql-listen.html>`_
-        registrations are removed, all session configuration
-        variables are reset to their default values, and all advisory locks
-        are released.
+        registrations are removed, all temporary tables are dropped,
+        all session configuration variables are reset to their default
+        values, and all advisory locks are released.
 
         Note that the above describes the default query returned by
         :meth:`Connection.get_reset_query`.  If one overloads the method
@@ -1781,6 +1781,8 @@ def get_reset_query(self):
             _reset_query.append('CLOSE ALL;')
         if caps.notifications and caps.plpgsql:
             _reset_query.append('UNLISTEN *;')
+        if caps.sql_discard_temp:
+            _reset_query.append('DISCARD TEMP;')
         if caps.sql_reset:
             _reset_query.append('RESET ALL;')
 
@@ -2724,7 +2726,8 @@ class _ConnectionProxy:
 ServerCapabilities = collections.namedtuple(
     'ServerCapabilities',
     ['advisory_locks', 'notifications', 'plpgsql', 'sql_reset',
-     'sql_close_all', 'sql_copy_from_where', 'jit'])
+     'sql_close_all', 'sql_copy_from_where', 'jit', 'sql_discard_temp'],
+    defaults=[False])
 ServerCapabilities.__doc__ = 'PostgreSQL server capabilities.'
 
 
@@ -2738,6 +2741,7 @@ def _detect_server_capabilities(server_version, connection_settings):
         sql_close_all = False
         jit = False
         sql_copy_from_where = False
+        sql_discard_temp = False
     elif hasattr(connection_settings, 'crdb_version'):
         # CockroachDB detected.
         advisory_locks = False
@@ -2747,6 +2751,7 @@ def _detect_server_capabilities(server_version, connection_settings):
         sql_close_all = False
         jit = False
         sql_copy_from_where = False
+        sql_discard_temp = False
     elif hasattr(connection_settings, 'crate_version'):
         # CrateDB detected.
         advisory_locks = False
@@ -2756,6 +2761,7 @@ def _detect_server_capabilities(server_version, connection_settings):
         sql_close_all = False
         jit = False
         sql_copy_from_where = False
+        sql_discard_temp = False
     else:
         # Standard PostgreSQL server assumed.
         advisory_locks = True
@@ -2765,6 +2771,7 @@ def _detect_server_capabilities(server_version, connection_settings):
         sql_close_all = True
         jit = server_version >= (11, 0)
         sql_copy_from_where = server_version.major >= 12
+        sql_discard_temp = True
 
     return ServerCapabilities(
         advisory_locks=advisory_locks,
@@ -2774,6 +2781,7 @@ def _detect_server_capabilities(server_version, connection_settings):
         sql_close_all=sql_close_all,
         sql_copy_from_where=sql_copy_from_where,
         jit=jit,
+        sql_discard_temp=sql_discard_temp,
     )
 
 

asyncpg/pool.py

@@ -1329,6 +1329,7 @@ def create_pool(dsn=None, *,
             SELECT pg_advisory_unlock_all();
             CLOSE ALL;
             UNLISTEN *;
+            DISCARD TEMP;
             RESET ALL;
 
         The exact reset query is determined by detected server capabilities,

tests/test_pool.py

@@ -605,6 +605,24 @@ async def get_xact_id(con):
                 id3 = await get_xact_id(con)
                 self.assertNotEqual(id2, id3)
 
+    async def test_pool_release_drops_temp_tables(self):
+        """Test that Connection.reset() drops temporary tables."""
+        async with self.create_pool(database='postgres',
+                                    init_size=1,
+                                    min_size=1,
+                                    max_size=1) as pool:
+            async with pool.acquire() as con:
+                real_con = con._con  # unwrap PoolConnectionProxy
+                await con.execute('CREATE TEMPORARY TABLE tmp_t (a int)')
+                await con.execute('INSERT INTO tmp_t VALUES (1)')
+
+            async with pool.acquire() as con:
+                self.assertIs(con._con, real_con)
+                self.assertIsNone(
+                    await con.fetchval("SELECT to_regclass('tmp_t')"))
+                # the name is free for the next user
+                await con.execute('CREATE TEMPORARY TABLE tmp_t (a int)')
+
     async def test_pool_connection_methods(self):
         async def test_fetch(pool):
             i = random.randint(0, 20)