Invalidate type cache on schema changes affecting statement result.

It is a logical continuation of the commit
749d857

Previously if a query used a composite type and it has changed then asyncpg
raised either IndexError or RuntimeError with no easy way to recover from it.

Unlike cases like in the mentioned commit here we can't reprepare and resend
the query because we fail at the very end in attempt to create a Record from
the response. It means the query is already done at the server's side.

Everything we can do in such case is to invalidate caches and mark the failed
prepared statement as "closed", it prevents getting it from the connection's
statement cache and the next query will be reprepared.
After all we have to reraise the exception to the caller.

For handling schema changes a new Connection.reload_schema_state() is added.
Now this method invalidates internal cached objects after changes in DB's
schema. In future it can do more actions to reflect schema changes to
internal representation.
It is intended for being called from callbacks for NOTIFY generated
from EVENT TRIGGERs or for using in migrations.

The other change is replacing RuntimeError=>OutdatedSchemaCacheError
when types in a composite type are not equal:
1. it is easier to catch;
2. it is easier to parse because now there is the composite type's name appear.

It can be considered as affecting compatibility with previous releases.

Author: vitaly-burovoy

asyncpg/connection.py

@@ -1236,6 +1236,51 @@ def _drop_global_statement_cache(self):
         else:
             self._drop_local_statement_cache()
 
+    async def reload_schema_state(self):
+        """Indicate that the database schema information must be reloaded.
+
+        For performance reasons, asyncpg caches certain aspects of the
+        database schema, such as the layout of composite types.  Consequently,
+        when the database schema changes, and asyncpg is not able to
+        gracefully recover from an error caused by outdated schema
+        assumptions, an :exc:`~asyncpg.exceptions.OutdatedSchemaCacheError`
+        is raised.  To prevent the exception, this method may be used to inform
+        asyncpg that the database schema has changed.
+
+        Example:
+
+        .. code-block:: pycon
+
+            >>> import asyncpg
+            >>> import asyncio
+            >>> async def change_type(con):
+            ...     result = await con.fetch('SELECT id, info FROM tbl')
+            ...     # Change composite's attribute type "int"=>"text"
+            ...     await con.execute('ALTER TYPE custom DROP ATTRIBUTE y')
+            ...     await con.execute('ALTER TYPE custom ADD ATTRIBUTE y text')
+            ...     await con.reload_schema_state()
+            ...     for id_, info in result:
+            ...         new = (info['x'], str(info['y']))
+            ...         await con.execute(
+            ...             'UPDATE tbl SET info=$2 WHERE id=$1', id_, new)
+            ...
+            >>> async def run():
+            ...     # Initial schema:
+            ...     # CREATE TYPE custom AS (x int, y int);
+            ...     # CREATE TABLE tbl(id int, info custom);
+            ...     con = await asyncpg.connect(user='postgres')
+            ...     async with con.transaction():
+            ...         # Prevent concurrent changes in the table
+            ...         await con.execute('LOCK TABLE tbl')
+            ...         await change_type(con)
+            ...
+            >>> asyncio.get_event_loop().run_until_complete(run())
+        """
+        # It is enough to clear the type cache only once, not in each
+        # connection in the pool.
+        self._protocol.get_settings().clear_type_cache()
+        self._drop_global_statement_cache()
+
     async def _execute(self, query, args, limit, timeout, return_status=False):
         with self._stmt_exclusive_section:
             result, _ = await self.__execute(
@@ -1277,6 +1322,15 @@ async def _do_execute(self, query, executor, timeout, retry=True):
                     after = time.monotonic()
                     timeout -= after - before
 
+        except exceptions.OutdatedSchemaCacheError:
+            # This exception is raised when we detect a difference between
+            # cached type's info and incoming tuple from the DB (when a type is
+            # changed by the ALTER TYPE).
+            # It is not possible to recover (the statement is already done at
+            # the server's side), the only way is to drop our caches and
+            # reraise the exception to the caller.
+            await self.reload_schema_state()
+            raise
         except exceptions.InvalidCachedStatementError:
             # PostgreSQL will raise an exception when it detects
             # that the result type of the query has changed from
@@ -1516,7 +1570,8 @@ def get(self, query, *, promote=True):
 
         if entry._statement.closed:
             # Happens in unittests when we call `stmt._state.mark_closed()`
-            # manually.
+            # manually or when a prepared statement closes itself on type
+            # cache error.
             self._entries.pop(query)
             self._clear_entry_callback(entry)
             return

asyncpg/exceptions/_base.py

@@ -12,7 +12,7 @@
 
 __all__ = ('PostgresError', 'FatalPostgresError', 'UnknownPostgresError',
            'InterfaceError', 'InterfaceWarning', 'PostgresLogMessage',
-           'InternalClientError')
+           'InternalClientError', 'OutdatedSchemaCacheError')
 
 
 def _is_asyncpg_class(cls):
@@ -221,6 +221,16 @@ class InternalClientError(Exception):
     pass
 
 
+class OutdatedSchemaCacheError(InternalClientError):
+    """A value decoding error caused by a schema change before row fetching."""
+
+    def __init__(self, msg, *, schema=None, data_type=None, position=None):
+        super().__init__(msg)
+        self.schema_name = schema
+        self.data_type_name = data_type
+        self.position = position
+
+
 class PostgresLogMessage(PostgresMessage):
     """A base class for non-error server messages."""
 

asyncpg/prepared_stmt.py

@@ -198,8 +198,17 @@ async def fetchrow(self, *args, timeout=None):
 
     async def __bind_execute(self, args, limit, timeout):
         protocol = self._connection._protocol
-        data, status, _ = await protocol.bind_execute(
-            self._state, args, '', limit, True, timeout)
+        try:
+            data, status, _ = await protocol.bind_execute(
+                self._state, args, '', limit, True, timeout)
+        except exceptions.OutdatedSchemaCacheError:
+            await self._connection.reload_schema_state()
+            # We can not find all manually created prepared statements, so just
+            # drop known cached ones in the `self._connection`.
+            # Other manually created prepared statements will fail and
+            # invalidate themselves (unfortunately, clearing caches again).
+            self._state.mark_closed()
+            raise
         self._last_status = status
         return data
 

asyncpg/protocol/codecs/base.pyx

@@ -5,6 +5,9 @@
 # the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0
 
 
+from asyncpg.exceptions import OutdatedSchemaCacheError
+
+
 cdef void* binary_codec_map[(MAXSUPPORTEDOID + 1) * 2]
 cdef void* text_codec_map[(MAXSUPPORTEDOID + 1) * 2]
 cdef dict TYPE_CODECS_CACHE = {}
@@ -190,20 +193,34 @@ cdef class Codec:
             FastReadBuffer elem_buf = FastReadBuffer.new()
 
         elem_count = <ssize_t><uint32_t>hton.unpack_int32(buf.read(4))
+        if elem_count != len(self.element_type_oids):
+            raise OutdatedSchemaCacheError(
+                'unexpected number of attributes of composite type: '
+                '{}, expected {}'
+                    .format(
+                        elem_count,
+                        len(self.element_type_oids),
+                    ),
+                schema=self.schema,
+                data_type=self.name,
+            )
         result = record.ApgRecord_New(self.element_names, elem_count)
         for i in range(elem_count):
             elem_typ = self.element_type_oids[i]
             received_elem_typ = <uint32_t>hton.unpack_int32(buf.read(4))
 
             if received_elem_typ != elem_typ:
-                raise RuntimeError(
+                raise OutdatedSchemaCacheError(
                     'unexpected data type of composite type attribute {}: '
                     '{!r}, expected {!r}'
                         .format(
                             i,
                             TYPEMAP.get(received_elem_typ, received_elem_typ),
                             TYPEMAP.get(elem_typ, elem_typ)
-                        )
+                        ),
+                    schema=self.schema,
+                    data_type=self.name,
+                    position=i,
                 )
 
             elem_len = hton.unpack_int32(buf.read(4))

asyncpg/protocol/settings.pxd

@@ -22,6 +22,7 @@ cdef class ConnectionSettings:
         decoder, format)
     cpdef inline remove_python_codec(
         self, typeoid, typename, typeschema)
+    cpdef inline clear_type_cache(self)
     cpdef inline set_builtin_type_codec(
         self, typeoid, typename, typeschema, typekind, alias_to)
     cpdef inline Codec get_data_codec(self, uint32_t oid, ServerDataFormat format=*)

asyncpg/protocol/settings.pyx

@@ -60,6 +60,9 @@ cdef class ConnectionSettings:
     cpdef inline remove_python_codec(self, typeoid, typename, typeschema):
         self._data_codecs.remove_python_codec(typeoid, typename, typeschema)
 
+    cpdef inline clear_type_cache(self):
+        self._data_codecs.clear_type_cache()
+
     cpdef inline set_builtin_type_codec(self, typeoid, typename, typeschema,
                                         typekind, alias_to):
         self._data_codecs.set_builtin_type_codec(typeoid, typename, typeschema,