test_cache_invalidation.py

# Copyright (C) 2016-present the asyncpg authors and contributors
# <see AUTHORS file>
#
# This module is part of asyncpg and is released under
# the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0


importasyncpg
fromasyncpgimport_testbaseastb

ERRNUM='unexpected number of attributes of composite type'
ERRTYP='unexpected data type of composite type'


classTestCacheInvalidation(tb.ConnectedTestCase):

def_get_cached_statements(self, connection=None):
ifconnectionisNone:
connection=self.con
returnlist(connection._stmt_cache.iter_statements())

def_check_statements_are_not_closed(self, statements):
self.assertGreater(len(statements), 0)
self.assertTrue(all(nots.closedforsinstatements))

def_check_statements_are_closed(self, statements):
self.assertGreater(len(statements), 0)
self.assertTrue(all(s.closedforsinstatements))

asyncdeftest_prepare_cache_invalidation_silent(self):
awaitself.con.execute('CREATE TABLE tab1(a int, b int)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, 2)')
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, 2))

statements=self._get_cached_statements()
self._check_statements_are_not_closed(statements)

awaitself.con.execute(
'ALTER TABLE tab1 ALTER COLUMN b SET DATA TYPE text')

result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, '2'))

self._check_statements_are_closed(statements)
finally:
awaitself.con.execute('DROP TABLE tab1')

asyncdeftest_prepare_cache_invalidation_in_transaction(self):
awaitself.con.execute('CREATE TABLE tab1(a int, b int)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, 2)')
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, 2))

statements=self._get_cached_statements()
self._check_statements_are_not_closed(statements)

awaitself.con.execute(
'ALTER TABLE tab1 ALTER COLUMN b SET DATA TYPE text')

withself.assertRaisesRegex(asyncpg.InvalidCachedStatementError,
'cached statement plan is invalid'):
asyncwithself.con.transaction():
result=awaitself.con.fetchrow('SELECT * FROM tab1')

self._check_statements_are_closed(statements)

# This is now OK,
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, '2'))
finally:
awaitself.con.execute('DROP TABLE tab1')

asyncdeftest_prepare_cache_invalidation_in_pool(self):
pool=awaitself.create_pool(database='postgres',
min_size=2, max_size=2)

awaitself.con.execute('CREATE TABLE tab1(a int, b int)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, 2)')

con1=awaitpool.acquire()
con2=awaitpool.acquire()

result=awaitcon1.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, 2))

result=awaitcon2.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, 2))

statements1=self._get_cached_statements(con1)
self._check_statements_are_not_closed(statements1)

statements2=self._get_cached_statements(con2)
self._check_statements_are_not_closed(statements2)

awaitself.con.execute(
'ALTER TABLE tab1 ALTER COLUMN b SET DATA TYPE text')

# con1 tries the same plan, will invalidate the cache
# for the entire pool.
result=awaitcon1.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, '2'))

self._check_statements_are_closed(statements1)
self._check_statements_are_closed(statements2)

asyncwithcon2.transaction():
# This should work, as con1 should have invalidated
# the plan cache.
result=awaitcon2.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, '2'))

finally:
awaitself.con.execute('DROP TABLE tab1')
awaitpool.release(con2)
awaitpool.release(con1)
awaitpool.close()

asyncdeftest_type_cache_invalidation_in_transaction(self):
awaitself.con.execute('CREATE TYPE typ1 AS (x int, y int)')
awaitself.con.execute('CREATE TABLE tab1(a int, b typ1)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, (2, 3))')
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3)))

statements=self._get_cached_statements()
self._check_statements_are_not_closed(statements)

asyncwithself.con.transaction():
awaitself.con.execute('ALTER TYPE typ1 ADD ATTRIBUTE c text')
withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitself.con.fetchrow('SELECT * FROM tab1')

self._check_statements_are_closed(statements)

# The second request must be correct (cache was dropped):
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3, None)))

# This is now OK, the cache is actual after the transaction.
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3, None)))
finally:
awaitself.con.execute('DROP TABLE tab1')
awaitself.con.execute('DROP TYPE typ1')

asyncdeftest_type_cache_invalidation_in_cancelled_transaction(self):
awaitself.con.execute('CREATE TYPE typ1 AS (x int, y int)')
awaitself.con.execute('CREATE TABLE tab1(a int, b typ1)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, (2, 3))')
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3)))

statements=self._get_cached_statements()
self._check_statements_are_not_closed(statements)

try:
asyncwithself.con.transaction():
awaitself.con.execute(
'ALTER TYPE typ1 ADD ATTRIBUTE c text')
withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitself.con.fetchrow('SELECT * FROM tab1')

self._check_statements_are_closed(statements)

# The second request must be correct (cache was dropped):
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3, None)))
raiseUserWarning# Just to generate ROLLBACK
exceptUserWarning:
pass

withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitself.con.fetchrow('SELECT * FROM tab1')
# This is now OK, the cache is filled after being dropped.
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3)))
finally:
awaitself.con.execute('DROP TABLE tab1')
awaitself.con.execute('DROP TYPE typ1')

asyncdeftest_prepared_type_cache_invalidation(self):
awaitself.con.execute('CREATE TYPE typ1 AS (x int, y int)')
awaitself.con.execute('CREATE TABLE tab1(a int, b typ1)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, (2, 3))')
prep=awaitself.con._prepare('SELECT * FROM tab1',
use_cache=True)
result=awaitprep.fetchrow()
self.assertEqual(result, (1, (2, 3)))

statements=self._get_cached_statements()
self._check_statements_are_not_closed(statements)

try:
asyncwithself.con.transaction():
awaitself.con.execute(
'ALTER TYPE typ1 ADD ATTRIBUTE c text')
withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitprep.fetchrow()

self._check_statements_are_closed(statements)

# PS has its local cache for types codecs, even after the
# cache cleanup it is not possible to use it.
# That's why it is marked as closed.
withself.assertRaisesRegex(
asyncpg.InterfaceError,
'the prepared statement is closed'):
awaitprep.fetchrow()

prep=awaitself.con._prepare('SELECT * FROM tab1',
use_cache=True)
# The second PS must be correct (cache was dropped):
result=awaitprep.fetchrow()
self.assertEqual(result, (1, (2, 3, None)))
raiseUserWarning# Just to generate ROLLBACK
exceptUserWarning:
pass

withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitprep.fetchrow()

# Reprepare it again after dropping cache.
prep=awaitself.con._prepare('SELECT * FROM tab1',
use_cache=True)
# This is now OK, the cache is filled after being dropped.
result=awaitprep.fetchrow()
self.assertEqual(result, (1, (2, 3)))
finally:
awaitself.con.execute('DROP TABLE tab1')
awaitself.con.execute('DROP TYPE typ1')

asyncdeftest_type_cache_invalidation_on_drop_type_attr(self):
awaitself.con.execute('CREATE TYPE typ1 AS (x int, y int, c text)')
awaitself.con.execute('CREATE TABLE tab1(a int, b typ1)')

try:
awaitself.con.execute(
'INSERT INTO tab1 VALUES (1, (2, 3, $1))', 'x')
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3, 'x')))

statements=self._get_cached_statements()
self._check_statements_are_not_closed(statements)

awaitself.con.execute('ALTER TYPE typ1 DROP ATTRIBUTE x')
withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitself.con.fetchrow('SELECT * FROM tab1')

self._check_statements_are_closed(statements)

# This is now OK, the cache is filled after being dropped.
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (3, 'x')))

finally:
awaitself.con.execute('DROP TABLE tab1')
awaitself.con.execute('DROP TYPE typ1')

asyncdeftest_type_cache_invalidation_on_change_attr(self):
awaitself.con.execute('CREATE TYPE typ1 AS (x int, y int)')
awaitself.con.execute('CREATE TABLE tab1(a int, b typ1)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, (2, 3))')
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3)))

statements=self._get_cached_statements()
self._check_statements_are_not_closed(statements)

# It is slightly artificial, but can take place in transactional
# schema changing. Nevertheless, if the code checks and raises it
# the most probable reason is a difference with the cache type.
awaitself.con.execute('ALTER TYPE typ1 DROP ATTRIBUTE y')
awaitself.con.execute('ALTER TYPE typ1 ADD ATTRIBUTE y bigint')
withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRTYP):
awaitself.con.fetchrow('SELECT * FROM tab1')

self._check_statements_are_closed(statements)

# This is now OK, the cache is filled after being dropped.
result=awaitself.con.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, None)))

finally:
awaitself.con.execute('DROP TABLE tab1')
awaitself.con.execute('DROP TYPE typ1')

asyncdeftest_type_cache_invalidation_in_pool(self):
awaitself.con.execute('CREATE DATABASE testdb')
pool=awaitself.create_pool(database='postgres',
min_size=2, max_size=2)

pool_chk=awaitself.create_pool(database='testdb',
min_size=2, max_size=2)

awaitself.con.execute('CREATE TYPE typ1 AS (x int, y int)')
awaitself.con.execute('CREATE TABLE tab1(a int, b typ1)')

try:
awaitself.con.execute('INSERT INTO tab1 VALUES (1, (2, 3))')

con1=awaitpool.acquire()
con2=awaitpool.acquire()

result=awaitcon1.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3)))

statements1=self._get_cached_statements(con1)
self._check_statements_are_not_closed(statements1)

result=awaitcon2.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3)))

statements2=self._get_cached_statements(con2)
self._check_statements_are_not_closed(statements2)

# Create the same schema in the "testdb", fetch data which caches
# type info.
con_chk=awaitpool_chk.acquire()
awaitcon_chk.execute('CREATE TYPE typ1 AS (x int, y int)')
awaitcon_chk.execute('CREATE TABLE tab1(a int, b typ1)')
awaitcon_chk.execute('INSERT INTO tab1 VALUES (1, (2, 3))')
result=awaitcon_chk.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3)))

statements_chk=self._get_cached_statements(con_chk)
self._check_statements_are_not_closed(statements_chk)

# Change schema in the databases.
awaitself.con.execute('ALTER TYPE typ1 ADD ATTRIBUTE c text')
awaitcon_chk.execute('ALTER TYPE typ1 ADD ATTRIBUTE c text')

# con1 tries to get cached type info, fails, but invalidates the
# cache for the entire pool.
withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitcon1.fetchrow('SELECT * FROM tab1')

self._check_statements_are_closed(statements1)
self._check_statements_are_closed(statements2)

asyncwithcon2.transaction():
# This should work, as con1 should have invalidated all caches.
result=awaitcon2.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3, None)))

# After all the con1 uses actual info from renewed cache entry.
result=awaitcon1.fetchrow('SELECT * FROM tab1')
self.assertEqual(result, (1, (2, 3, None)))

# Check the invalidation is database-specific, i.e. cache entries
# for pool_chk/con_chk was not dropped via pool/con1.

self._check_statements_are_not_closed(statements_chk)

withself.assertRaisesRegex(
asyncpg.OutdatedSchemaCacheError, ERRNUM):
awaitcon_chk.fetchrow('SELECT * FROM tab1')

self._check_statements_are_closed(statements_chk)
finally:
awaitself.con.execute('DROP TABLE tab1')
awaitself.con.execute('DROP TYPE typ1')
awaitpool.release(con2)
awaitpool.release(con1)
awaitpool.close()
awaitpool_chk.release(con_chk)
awaitpool_chk.close()
awaitself.con.execute('DROP DATABASE testdb')