Asyncpg: Connection.reset fails on Amazon Redshift

[emirot]

Let's say I have two long running queries that I'd like to run concurrently
Here is my questions :

• Does asyncpg will help in that case ?
• using concurrent.futures will do the same ?

import asyncio
import asyncpg

async def connect_to_db():
    pool = await asyncpg.create_pool("postgres://user:pass@localhost:5555/dev",
                                     command_timeout=60)
    conn1 = await pool.acquire()
    conn2 = await pool.acquire()
    return pool, conn1, conn2

async def create_table_a(conn):
    await conn.execute('CREATE TABLE my_schema.mytab_a (a int)')

async def create_table_b(conn):
    await conn.execute('CREATE TABLE my_schema.mytab_b (b int)')

loop = asyncio.get_event_loop()
pool, conn1, conn2 = loop.run_until_complete(connect_to_db())

tasks = [asyncio.ensure_future(create_table_a(conn1)), asyncio.ensure_future(create_table_b(conn2))]
loop.run_until_complete(asyncio.wait(tasks))
#release the pool and the connections

Thanks for this library

[elprans]

Sure, if you use separate connections (which you are), it is possible to run an arbitrary number of queries concurrently. That's basically the main feature of asyncpg :-).

Now, whether or not the queries _actually_ execute concurrently depends on the locking that your queries cause. For example, if you run ALTER TABLE on the same table concurrently, the execution will actually be sequential since ALTER TABLE locks the whole table on Postgres side.

Thanks for this library

Cheers!

[emirot]

Thanks a lot @elprans for you response.
I came across an issue using the simple example from the documentation :

import asyncio
import asyncpg

async def connect_to_db():
    pool = await asyncpg.create_pool("postgres://user:pass@localhost:5555/dev",
                                     command_timeout=60)
    conn1 = await pool.acquire()
    try:
        a = await conn1.fetch("""SELECT 1;""")
        print(a)
    finally:
        await pool.release(conn1)
loop = asyncio.get_event_loop()
loop.run_until_complete(connect_to_db())

Here is the stacktrace :

[<Record ?column?=1>]
Traceback (most recent call last):
  File "test.py", line 18, in <module>
    loop.run_until_complete(connect_to_db())
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
    return future.result()
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/asyncio/tasks.py", line 241, in _step
    result = coro.throw(exc)
  File "test.py", line 14, in connect_to_db
    await pool.release(conn1)
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/site-packages/asyncpg/pool.py", line 184, in release
    await connection.reset()
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/site-packages/asyncpg/connection.py", line 329, in reset
    ''')
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/site-packages/asyncpg/connection.py", line 120, in execute
    return await self._protocol.query(script, timeout)
  File "asyncpg/protocol/protocol.pyx", line 224, in query (asyncpg/protocol/protocol.c:46945)
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/asyncio/futures.py", line 361, in __iter__
    yield self  # This tells Task to wait for completion.
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/asyncio/tasks.py", line 296, in _wakeup
    future.result()
  File "/Users/nolanemirot/miniconda/envs/pyt35/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
asyncpg.exceptions.PostgresSyntaxError: syntax error at or near "ALL"

Looking at the source code the error https://github.com/MagicStack/asyncpg/blob/master/asyncpg/exceptions/__init__.py#L554 I don't understand the relation with a postgres syntax error. (I'm connecting to a redshift cluster)
Replacing await pool.release(conn1) with pool.close() doesn't trigger any error.
Do you have any idea why ?

[elprans]

This is triggered by Connection.reset(): https://github.com/MagicStack/asyncpg/blob/master/asyncpg/connection.py#L349

It is likely that Redshift's version of postgres doesn't support either CLOSE ALL or RESET ALL. Which version is running there?

[emirot]

Calling SELECT VERSION(); gives me
PostgreSQL 8.0.2 on i686-pc-linux-gnu, compiled by GCC gcc (GCC) 3.4.2 20041017 (Red Hat 3.4.2-6.fc3), Redshift 1.0.1096

[elprans]

Ouch! That's ancient! That version was released over 10 years ago.

[emirot]

Yes it is: https://docs.aws.amazon.com/redshift/latest/dg/c_redshift-and-postgres-sql.html

[elprans]

OK, we'll need to find a way to close the cursors on that version. Meanwhile, you can just drop the CLOSE ALL; line from that statement.

[sintezcs]

Are there any updates to this issue? I've tried the proposed workaround (drop the CLOSE ALL;) but it doesn't help.

[mdespriee]

Interested in asyncpg for Redshift too.
@elprans @emirot Just to understand : asyncpg does not work at all with redshift, or only connection pooling ?

[elprans]

We'll probably be able to get some level of support going. I'll try to take a look soon.