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 

[SlavaSkvortsov]

We had a similar problem due to multiple web workers. They generated prepared statements with the same names - original function to generate IDs looks like this:

    def _get_unique_id(self, prefix):
        global _uid
        _uid += 1
        return '__asyncpg_{}_{:x}__'.format(prefix, _uid)

So we just changed the Connection class a bit

from uuid import uuid4

from asyncpg import Connection


class CConnection(Connection):
    def _get_unique_id(self, prefix: str) -> str:
        return f'__asyncpg_{prefix}_{uuid4()}__'

You need to provide it when you create the engine

engine = create_async_engine(
    settings.database_url,
    connect_args={
        'connection_class': CConnection,
    },
)

[CaselIT]

@SlavaSkvortsov that seems to be something we cannot change on out end, other than subclassing the connection class to change a private method like in you example.

You may want to open an issue on the asyncpg repo for a solution out of the box. Imho _get_unique_id should at least take into consideration the process pid (and/or thread id). The best way would be the ability of customizing it, maybe using a connect kwarg or with the ability to specify a name when calling prepare, since there is no way of manually assigning a name to a statement

[forshev]

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

Should this configuration pass the prepared statement cache size parameter to asyncpg connection?
With SQLAlchemy@1.4.22 it doesn't.

So i managed to set cache size on asyncpg side like this:

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@hostname/dbname?prepared_statement_cache_size=0",
    poolclass=NullPool,
    future=True,
    connect_args={'statement_cache_size': 0},
)

Note the connect_args={'statement_cache_size': 0} parameter.

And in combination with @SlavaSkvortsov's unique ids suggestion it seems that i got rid of prepared statement errors with pgbouncer.

[rslinckx]

When using asyncpg only, you can use statement_cache_size=0 and it won't use prepared statements at all, thus working behind pgbouncer in transaction mode.

My understanding is that sqlalchemy/asyncpg will use prepared statements no matter what the setting prepared_statement_cache_size value. This means that it will create prepared statements in any case. Using the unique ID workaround only hides the problem, you will create a lot of prepared statements in your open sessions to the database (each with a unique name), and each transaction will get a random session including some subset of those statements. Since you are disabling the cache, it won't be a real issue as it will keep creating new ones, and they will never conflict because the name is random, but they are still there and created. I'm not sure what's the mechanism to delete them, maybe they just get deleted whenever the backend session is closed, once in a while?

As for why sqlalchemy needs to use prepared statements in any case, I have no idea.

[CaselIT]

As for why sqlalchemy needs to use prepared statements in any case, I have no idea.

asyncpg has no other way of returning the type information of the selected columns in the query when using a normal query, so sqlalchemy needs to use prepared statements:

sqlalchemy/lib/sqlalchemy/dialects/postgresql/asyncpg.py

Lines 613 to 616 in 990eb3d

	if cache is None:
	prepared_stmt = await self._connection.prepare(operation)
	attributes = prepared_stmt.get_attributes()
	return prepared_stmt, attributes

and

sqlalchemy/lib/sqlalchemy/dialects/postgresql/asyncpg.py

Lines 392 to 404 in 990eb3d

	if attributes:
	self.description = [
	(
	attr.name,
	attr.type.oid,
	None,
	None,
	None,
	None,
	None,
	)
	for attr in attributes
	]

[zzzeek]

As for why sqlalchemy needs to use prepared statements in any case, I have no idea.

#6467 (comment)

[zzzeek]

assuming we keep using prepared statements theres a bunch of people on this issue now can we come up with a way to fix what's wrong here and get this closed? thanks

[zzzeek]

MagicStack/asyncpg#837 is closed via MagicStack/asyncpg#846 . we will want to expose this and then add a documentation section that this is a means of using our asyncpg driver with pgbouncer along with a recipe based on UUID or similar.

[jacksund]

Is there any update on how use sqlalchemy+asyncpg+pgbouncer? I'm new to sqlalchemy's async engine and stumbled onto this issue while trying out Prefect's v2 beta. Their package uses create_async_engine under the hood, but even if I modified their implementation, I don't think they would accept the strategy used by @SlavaSkvortsov. The only other suggestion I've seen is setting pgbouncer’s pool_mode to session (according to asyncpg's FAQ), but I didn't have any luck with it.

EDIT: I should add -- session mode didn't work for me because I have many long-lived clients connecting. So I depend on a pool in transaction mode.

[zzzeek]

yes something like that. but i dont think it should be necessary, because once the transaction is released back to pgbouncer, that prepared statement is useless anyway, so just DEALLOCATE them all, at the start of a transaction is best. then, we also can't rely on any kind of caching of these prepared statements, because with transactional mode, they are similarly lost every transaction to us, hence prepared statement cache needs to be zero. those two steps should solve the problem. if not, then i dont understand what's going on.

[zzzeek]

we could probably support statement mode also if the engine is run with autocommit

[zzzeek]

confirm the prepared_statement_cache_size works setting it to zero, test case:

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine


async def async_main():
    engine = create_async_engine(
        "postgresql+asyncpg://scott:tiger@localhost/test",
        echo=True,

        # when commenting this out, we get
        # "prepared statement "__asyncpg_stmt_7__" does not exist"
        # setting the cache to zero allows the below statement to invoke
        connect_args=dict(prepared_statement_cache_size=0)
    )

    for i in range(3):
        async with engine.begin() as conn:
            await conn.execute(text("select 1"))
            await conn.exec_driver_sql("DEALLOCATE ALL")

asyncio.run(async_main())

[vamshiaruru-virgodesigns]

Just want to chime in and say that setting prepared_statement_cache_size to zero in create_async_engine doesn't help with prepared statement already exists error. The solution from comment #6467 (comment) works.

[zzzeek]

@vamshiaruru-virgodesigns can you confirm some points for me?

• are you using session or transaction mode in pgbouncer?
• confirm that Using statement_cache_size asyncpg setting / prepared statement name for asyncpg w pgbouncer #6467 (comment) does not work ? note I just updated it to include prepared_statement_cache_size=0

The solution of just naming them all randomly seems very much like it would fill up the PostgreSQL session with thousands of unused prepared statements which we would assume uses memory.

[vamshiaruru-virgodesigns]

@zzzeek , I'm using transaction mode in pgbouncer, and just setting prepared_statement_cache_size or statement_cache_size to zero still doesn't work. (I keep getting the prepared statement already exists error). Only after adding the custom Connector I was able to get SQLAlchemy working with PGBouncer.
I haven't noticed this block:

    @event.listens_for(engine.sync_engine, "begin")
    def clear_prepared_statements_on_begin(conn):
        conn.exec_driver_sql("DEALLOCATE ALL")

I can quickly try with that and see if it works without custom connector.

[zzzeek]

thanks for the quick reply. if you keep working with the other approach, can you check on the memory use of your postgresql workers, if you are doing things at scale. it intuitively seems like it would fill up memory but I don't really know the specifics.

[vamshiaruru-virgodesigns]

Update, with this code

engine = create_async_engine(
    settings.database_dsn,
    pool_size=settings.sqlalchemy_pool_size,
    pool_pre_ping=True,
    connect_args={
        "statement_cache_size": 0,
        "prepared_statement_cache_size": 0,
        # "connection_class": CConnection,
    },
)

@event.listens_for(engine.sync_engine, "begin")
def clear_prepared_statements_on_begin(conn):
    conn.exec_driver_sql("DEALLOCATE ALL")

It works. But I am not sure how expensive it is to run DEALLOCATE ALL at every begin.
A question I have is this. AsyncPG documentation says setting statement_cache_size to 0 disables prepared queries (doc link is here https://magicstack.github.io/asyncpg/current/faq.html?highlight=statement_cache_size ), but just setting that connection argument without using your DEALLOCATE ALL handler doesn't work. I am curious as to why that could be happening.
As for your memory query, I very recently implemented the change to use custom Connector, and in that time our memory usage hasn't increased. We usually have ~1.7k db connections on our db. But we'll have to keep monitoring it as its been only few hours since I've implemented it. But logically speaking, postgres automatically deallocates a prepared statement when session ends, so regardless of how we name the prepared statement, those would be released, right? I could be wrong here since I don't know asyncpg internals, but it feels like it shouldn't make a difference how we constructed the query name.

Here's how asyncpg constructs the name:

def _get_unique_id(self, prefix):
        global _uid
        _uid += 1
        return '__asyncpg_{}_{:x}__'.format(prefix, _uid)

I don't think they are keeping track of this _uid anywhere and it is just their way of making a unique name maybe?

[zzzeek]

Update, with this code

engine = create_async_engine(
    settings.database_dsn,
    pool_size=settings.sqlalchemy_pool_size,
    pool_pre_ping=True,
    connect_args={
        "statement_cache_size": 0,
        "prepared_statement_cache_size": 0,
        # "connection_class": CConnection,
    },
)

@event.listens_for(engine.sync_engine, "begin")
def clear_prepared_statements_on_begin(conn):
    conn.exec_driver_sql("DEALLOCATE ALL")

It works. But I am not sure how expensive it is to run DEALLOCATE ALL at every begin.

pgbouncer itself, when using session mode (which I still think is a much better idea here) uses DISCARD ALL by default which is more "heavy", although it does this after connection release so I suppose this is something it can do in the background on its end.

A question I have is this. AsyncPG documentation says setting statement_cache_size to 0 disables prepared queries (doc link is here https://magicstack.github.io/asyncpg/current/faq.html?highlight=statement_cache_size ),

it actually gives a hint:

(and, obviously, avoid the use of Connection.prepare());

we are using connection.prepare() for all statements. We have no choice in that regard due to limitations in asyncpg's API.

As for your memory query, I very recently implemented the change to use custom Connector, and in that time our memory usage hasn't increased. We usually have ~1.7k db connections on our db. But we'll have to keep monitoring it as its been only few hours since I've implemented it. But logically speaking, postgres automatically deallocates a prepared statement when session ends, so regardless of how we name the prepared statement, those would be released, right?

no, because PG Bouncer is pooling the connections, hence those sessions are still opened. per docs the server reset query isn't used. There is also server_reset_query_always which as they advertise is for "broken" apps. From my POV using PGBouncer transaction mode with asyncpg's prepared statements is "broken", I would agree. I dont see the wisdom at all in using a different connection for every transaction, when SQLAlchemy already has a clear system of distinguishing the scope of a "connection" from that of a "transaction" and should not be worked around.

Here's how asyncpg constructs the name:

def _get_unique_id(self, prefix):
        global _uid
        _uid += 1
        return '__asyncpg_{}_{:x}__'.format(prefix, _uid)

I don't think they are keeping track of this _uid anywhere and it is just their way of making a unique name maybe?

they dont need to make it globally unique because asyncpg assumes their connection corresponds to a single PostgreSQL session that did not exist beforehand.

[zzzeek]

I mean performancewise, the prepared_statement_cache_size=0 setting definitely degrades performance very measurably.

I really think someone should do some benches here with pgbouncer session vs. transaction mode, I think the former will perform better in every way. transaction mode seems like it is intended for applications that are mis-designed to hold onto single connections for long idle spans, or something (Edit: yes like when you still want to have client side pooling in place also. OK).

[zzzeek]

im actually going to document this. I see zero advantage at all to transaction pooling mode (Edit: OK it can reduce latency on connect by keeping QueuePool in place)

[vamshiaruru-virgodesigns]

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.

[zzzeek]

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.

[mimre25]

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.

[CaselIT]

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

[zzzeek]

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.

[mimre25]

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.

[zzzeek]

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 ?

[CaselIT]

it may make sense to move to a discussion?

[zzzeek]

I think so?

[CaselIT]

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