Correct the check for the use of unnamed statement by introspection

Author: elprans

asyncpg/connection.py

@@ -272,7 +272,7 @@ async def _get_statement(self, query, timeout, *, named: bool=False):
         # Only use the cache when:
         #  * `statement_cache_size` is greater than 0;
         #  * query size is less than `max_cacheable_statement_size`.
-        use_cache = cache_enabled = self._stmt_cache.get_max_size() > 0
+        use_cache = self._stmt_cache.get_max_size() > 0
         if (use_cache and
                 self._config.max_cacheable_statement_size and
                 len(query) > self._config.max_cacheable_statement_size):
@@ -286,13 +286,13 @@ async def _get_statement(self, query, timeout, *, named: bool=False):
         statement = await self._protocol.prepare(stmt_name, query, timeout)
         ready = statement._init_types()
         if ready is not True:
-            types = await self.__execute(self._intro_query, (list(ready),),
-                                         0, timeout)
+            types, intro_stmt = await self.__execute(
+                self._intro_query, (list(ready),), 0, timeout)
             self._protocol.get_settings().register_data_types(types)
-            if not cache_enabled:
-                # The execution of the introspection query with statement
-                # cache turned off has blown away the anonymous statement
-                # we've prepared for the query, so we need to re-prepare it.
+            if not intro_stmt.name and not statement.name:
+                # The introspection query has used an anonymous statement,
+                # which has blown away the anonymous statement we've prepared
+                # for the query, so we need to re-prepare it.
                 statement = await self._protocol.prepare(
                     stmt_name, query, timeout)
 
@@ -1199,8 +1199,9 @@ def _drop_global_statement_cache(self):
 
     async def _execute(self, query, args, limit, timeout, return_status=False):
         with self._stmt_exclusive_section:
-            return await self.__execute(query, args, limit, timeout,
-                                        return_status=return_status)
+            result, _ = await self.__execute(
+                query, args, limit, timeout, return_status=return_status)
+        return result
 
     async def __execute(self, query, args, limit, timeout,
                         return_status=False):
@@ -1214,7 +1215,8 @@ async def _executemany(self, query, args, timeout):
             stmt, args, '', timeout)
         timeout = self._protocol._get_timeout(timeout)
         with self._stmt_exclusive_section:
-            return await self._do_execute(query, executor, timeout)
+            result, _ = await self._do_execute(query, executor, timeout)
+        return result
 
     async def _do_execute(self, query, executor, timeout, retry=True):
         if timeout is None:
@@ -1263,10 +1265,10 @@ async def _do_execute(self, query, executor, timeout, retry=True):
             if self._protocol.is_in_transaction() or not retry:
                 raise
             else:
-                result = await self._do_execute(
+                return await self._do_execute(
                     query, executor, timeout, retry=False)
 
-        return result
+        return result, stmt
 
 
 async def connect(dsn=None, *,

tests/test_introspection.py

@@ -46,9 +46,7 @@ async def test_introspection_on_large_db(self):
             await self.con.fetchval('SELECT $1::int[]', [1, 2])
 
     @tb.with_connection_options(statement_cache_size=0)
-    async def test_introspection_no_stmt_cache(self):
-        # Check that type introspection does not use prepared statements
-        # when the statement cache is disabled.
+    async def test_introspection_no_stmt_cache_01(self):
         self.assertEqual(self.con._stmt_cache.get_max_size(), 0)
         await self.con.fetchval('SELECT $1::int[]', [1, 2])
 
@@ -63,3 +61,19 @@ async def test_introspection_no_stmt_cache(self):
             await self.con.execute('''
                 DROP EXTENSION hstore
             ''')
+
+    @tb.with_connection_options(max_cacheable_statement_size=1)
+    async def test_introspection_no_stmt_cache_02(self):
+        await self.con.fetchval('SELECT $1::int[]', [1, 2])
+
+        await self.con.execute('''
+            CREATE EXTENSION IF NOT EXISTS hstore
+        ''')
+
+        try:
+            await self.con.set_builtin_type_codec(
+                'hstore', codec_name='pg_contrib.hstore')
+        finally:
+            await self.con.execute('''
+                DROP EXTENSION hstore
+            ''')