Using statement_cache_size asyncpg setting / prepared statement name for asyncpg w pgbouncer

[dyens]

Hi!

I use sqlalchemy 1.4 with asyncpg driver with pgbouncer.

    from sqlalchemy.ext.asyncio import create_async_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.ext.asyncio import AsyncSession

    engine = create_async_engine(
        f'postgresql+asyncpg://{username}:{password}@{host}:{port}/{dbname}',
        echo=False,
    )
    session_maker = sessionmaker(
        engine,
        class_=AsyncSession,
    )

I have an error:

asyncpg.exceptions.DuplicatePreparedStatementError: prepared statement "__asyncpg_stmt_a__" already exists
HINT:
NOTE: pgbouncer with pool_mode set to "transaction" or
"statement" does not support prepared statements properly.
You have two options:

* if you are using pgbouncer for connection pooling to a
  single server, switch to the connection pool functionality
  provided by asyncpg, it is a much better option for this
  purpose;

* if you have no option of avoiding the use of pgbouncer,
  then you can set statement_cache_size to 0 when creating
  the asyncpg connection object.

How i can pass this setting (statement_cache_size=0) to asyncpg connection object?

[zzzeek]

SQLAlchemy doesn't make use of "statement_cache_size" as it necessarily uses the asyncpg.connection.prepare() method directly, so there is no way to disable the use of prepared statements across the board. However, you can remove the use of prepared statement caching itself, so that it will make a new prepared statement each time, using prepared_statement_cache_size=0, if you can try that and see if it works we can mention this in the docs:

https://docs.sqlalchemy.org/en/14/dialects/postgresql.html#prepared-statement-cache

[dyens]

Thank you, @zzzeek

In my case it does not work.

I found in :

https://github.com/sqlalchemy/sqlalchemy/blob/master/lib/sqlalchemy/dialects/postgresql/asyncpg.py#L747

                await_only(self.asyncpg.connect(*arg, **kw))

in **kw we does not pass statement_cache_size.

If i add:

if 'prepared_statement_cache_size' in kw:
    kw['statement_cache_size'] = kw['prepared_statement_cache_size']

It helps sometimes, but sometimes does not (i don't know why...)

[zzzeek]

hi -

did you try just direct usage of the prepared_statement_cache_size variable as given:

engine = create_async_engine("postgresql+asyncpg://user:pass@hostname/dbname?prepared_statement_cache_size=0")

? if that doesn't work, we are out of luck. SQLAlchemy is forced to use connection.prepare() which means we have to use prepared statements.

[dyens]

Yes, i use this:

engine = create_async_engine("postgresql+asyncpg://user:pass@hostname/dbname?prepared_statement_cache_size=0")

[zzzeek]

all of our SELECT statements have to use connection.prepare() because we need to be able to call get_attributes(). if asyncpg could be convinced to give us this API without necessitating the use of connection.prepare() we could begin to think about how to support that.

[zzzeek]

the doc at https://magicstack.github.io/asyncpg/current/api/index.html?highlight=statement_cache_size doesnt claim this disables prepared statements, just that it doesn't cache them:

statement_cache_size (int) – The size of prepared statement LRU cache. Pass 0 to disable the cache.

oh perhaps this is needed for when we do INSERT/UPDATE/DELETE, OK.... the patch you tried at kw['statement_cache_size'] = kw['prepared_statement_cache_size'] should do that.

[zzzeek]

that is, if you use prepared_statement_cache_size=0 and add code like this:

diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 4a191cd286..415862a17b 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -735,6 +735,7 @@ class AsyncAdapt_asyncpg_dbapi:
         prepared_statement_cache_size = kw.pop(
             "prepared_statement_cache_size", 100
         )
+        kw["statement_cache_size"] = prepared_statement_cache_size
         if util.asbool(async_fallback):
             return AsyncAdaptFallback_asyncpg_connection(
                 self,

that's the best we can do. if there are still problems then we need support from asyncpg

[dyens]

I try this. But for some reason this is not help. (sometimes an error appears but sometimes not...)

[zzzeek]

this might be related to the fact that we still use connection.prepare(). I dont know the internals of asyncpg enough to advise further on what might be going on.

[dyens]

Yes, this is a reason..

Also in AsyncAdapt_asyncpg_cursor._prepare_and_execute

We actively use prepared statements for executing queries.

In asyncpg, as i can see, for not prepared queries they does not use prepared statements

# Pseudo code:

# In asyncpg for statement_cache_size = 0
stmt = Connection._get_statement(
        self,
        query,
        named=False  # named=True is used for connection.prepare call
)

# in deep of this function, if named = False and statement_cache_size = 0
# asyncpg use empty name for prepared statements.

result = Connection._protocol.bind_execute(stmt, args, ...) # a little wrong but essence is