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

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?

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

@zzzeek

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...)

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.

Yes, i use this:

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

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.

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.

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

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

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.

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

Thanks for your explanations. In our requirements, we have a django application connecting to same DB and I have read that session pooling with django is not very performant compared to transaction level pooling. But I haven't personally performed any bench marks against it.

OK yes, thinking some more, using transaction mode, you can keep using QueuePool and that will save you on the connectivity overhead as well, if PGBouncer is able to keep many client connections persistently linked to a smaller set of PG server connections, OK, I get it then.

From all of this it seems like the best approach would be to use server_reset_query_always . that way the server reset query is kept out of the client, it's done on the server after the connection has been released.

I wanted to chime in here, as I recently faced this problem and solved an edge case that wasn't discussed above.

While the solutions from #6467 (comment) and #6467 (comment) work in most scenarios, there is a further problem, if one wants to use transaction isolation levels.
If you want to use a statement a la

SET TRANSACTION ISOLATION LEVEL <level>

then you need to run this as first statement of a connection.

In such a scenario the event listener from #6467 (comment) doesn't work, as it only rhttps://github.com//issues/6467#issuecomment-864943824uns after the BEGIN statement.

However, you can hook into the connect event of the engine's pool, as this is run right after the connection is created.:

async_engine = ...

@event.listens_for(async_engine.sync_engine.pool, "connect")
def clear_prepared_statements_on_begin_and_isolate(conn, branch):
    conn.run_async(
        lambda con: con.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;")
    )
    conn.run_async(lambda con: con.execute("DEALLOCATE ALL;"))

As I had the situation that the transaction isolation is only necessary in some cases, I've created this gist to allow acquiring a session with and without the isolation level.

Note that, if you use the uuid approach (#6467 (comment)) this is probably not necessary, but it was not an option in my scenario.

@mimre25

@mimre25 I don't think that gist works at all:

session: AsyncSession = sessionmaker(engine)()

this returns you a Session, not an AsyncSession. You should need to use async_sessionmaker with async sessions.
Also there is no need to create a session maker if you instantiate it inside a function every time it's called. Just instantiate an AsyncSession directly (a session maker is useful only if it's placed at the module level)

I wanted to chime in here, as I recently faced this problem and solved an edge case that wasn't discussed above.

While the solutions from #6467 (comment) and #6467 (comment) work in most scenarios, there is a further problem, if one wants to use transaction isolation levels. If you want to use a statement a la
SET TRANSACTION ISOLATION LEVEL <level>
then you need to run this as first statement of a connection.

In such a scenario the event listener from #6467 (comment) doesn't work, as it only rhttps://github.com//issues/6467#issuecomment-864943824uns after the BEGIN statement.

However, you can hook into the connect event of the engine's pool, as this is run right after the connection is created.:
async_engine = ...

@event.listens_for(async_engine.sync_engine.pool, "connect")
def clear_prepared_statements_on_begin_and_isolate(conn, branch):
    conn.run_async(
        lambda con: con.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;")
    )
    conn.run_async(lambda con: con.execute("DEALLOCATE ALL;"))
As I had the situation that the transaction isolation is only necessary in some cases, I've created this gist to allow acquiring a session with and without the isolation level.

Just so you know, for the isolation level part of the above, this feature is built in. Simply use engine-level isolation level. does the exact same thing.

@CaselIT

Thanks for pointing that out @CaselIT - I've corrected the gist to directly instantiate the AsyncSession.

@zzzeek thank you for that - I wasn't aware of that. - I won't edit the gist in this regard (besides adding a comment), as it still shows how to "hook" directly in the underlying connection.

Add name_func optional attribute for asyncpg adapter

why is this issue opened? you can use prepared statements with pgbouncer + asyncpg. it's working here right in
#10226 save for one issue with "ping" that we can fix.

Is there anyone here still unclear on how to use asyncpg with pgbouncer ?

it may make sense to move to a discussion?

I think so?

ok, so once fixed we can add a new post with the link to the fix

Navigation Menu

Repository navigation

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

41 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

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

Description

Activity

zzzeek commented on May 11, 2021

dyens commented on May 11, 2021

zzzeek commented on May 11, 2021

dyens commented on May 11, 2021

zzzeek commented on May 11, 2021

zzzeek commented on May 11, 2021

zzzeek commented on May 11, 2021

dyens commented on May 11, 2021

zzzeek commented on May 11, 2021

dyens commented on May 11, 2021

41 remaining items

vamshiaruru-virgodesigns commented on Aug 18, 2022

zzzeek commented on Aug 18, 2022

mimre25 commented on Feb 5, 2023

CaselIT commented on Feb 5, 2023

zzzeek commented on Feb 5, 2023

mimre25 commented on Feb 9, 2023

zzzeek commented on Aug 16, 2023

CaselIT commented on Aug 16, 2023

zzzeek commented on Aug 16, 2023

CaselIT commented on Aug 16, 2023

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions