From: Daniele Varrazzo Date: Wed, 8 Jun 2022 14:25:28 +0000 (+0200) Subject: fix: don't use PQsendQuery in pipeline mode X-Git-Tag: 3.1~64^2 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=c894d10cb10e7d68ced9e388a4d954c98108faa8;p=thirdparty%2Fpsycopg.git fix: don't use PQsendQuery in pipeline mode There is no reason to use it: the only reason to not use PQsendQueryParams is to send multiple statements, but this is not possible in pipeline mode anyway. Using PQsendQuery seems to produce a spurious Close message, which the libpq is surprised to receive. Issue reported to pgsql-bugs. https://www.postgresql.org/message-id/CA%2Bmi_8bvD0_CW3sumgwPvWdNzXY32itoG_16tDYRu_1S2gV2iw%40mail.gmail.com Close #314 --- diff --git a/psycopg/psycopg/client_cursor.py b/psycopg/psycopg/client_cursor.py index bdaa4e127..2a1ce3078 100644 --- a/psycopg/psycopg/client_cursor.py +++ b/psycopg/psycopg/client_cursor.py @@ -58,22 +58,19 @@ class ClientCursorMixin(BaseCursor[ConnectionType, Row]): ) self._query = query - if no_pqexec: - if self._conn._pipeline: - self._conn._pipeline.command_queue.append( - partial(self._pgconn.send_query_params, None) - ) - else: - self._pgconn.send_query_params(query.query, None) + + if self._conn._pipeline: + # In pipeline mode always use PQsendQueryParams - see #314 + # Multiple statements in the same query are not allowed anyway. + self._conn._pipeline.command_queue.append( + partial(self._pgconn.send_query_params, query.query, None) + ) + elif no_pqexec: + self._pgconn.send_query_params(query.query, None) else: # if we don't have to, let's use exec_ as it can run more than # one query in one go - if self._conn._pipeline: - self._conn._pipeline.command_queue.append( - partial(self._pgconn.send_query, query.query) - ) - else: - self._pgconn.send_query(query.query) + self._pgconn.send_query(query.query) def _convert_query( self, query: Query, params: Optional[Params] = None diff --git a/psycopg/psycopg/cursor.py b/psycopg/psycopg/cursor.py index a2777e470..4a0de6261 100644 --- a/psycopg/psycopg/cursor.py +++ b/psycopg/psycopg/cursor.py @@ -448,35 +448,32 @@ class BaseCursor(Generic[ConnectionType, Row]): fmt = BINARY if binary else TEXT self._query = query - if query.params or no_pqexec or fmt == BINARY: - if self._conn._pipeline: - self._conn._pipeline.command_queue.append( - partial( - self._pgconn.send_query_params, - query.query, - query.params, - param_formats=query.formats, - param_types=query.types, - result_format=fmt, - ) - ) - else: - self._pgconn.send_query_params( + + if self._conn._pipeline: + # In pipeline mode always use PQsendQueryParams - see #314 + # Multiple statements in the same query are not allowed anyway. + self._conn._pipeline.command_queue.append( + partial( + self._pgconn.send_query_params, query.query, query.params, param_formats=query.formats, param_types=query.types, result_format=fmt, ) + ) + elif no_pqexec or query.params or fmt == BINARY: + self._pgconn.send_query_params( + query.query, + query.params, + param_formats=query.formats, + param_types=query.types, + result_format=fmt, + ) else: # if we don't have to, let's use exec_ as it can run more than # one query in one go - if self._conn._pipeline: - self._conn._pipeline.command_queue.append( - partial(self._pgconn.send_query, query.query) - ) - else: - self._pgconn.send_query(query.query) + self._pgconn.send_query(query.query) def _convert_query( self, query: Query, params: Optional[Params] = None diff --git a/tests/test_client_cursor.py b/tests/test_client_cursor.py index dca40221d..e42993b71 100644 --- a/tests/test_client_cursor.py +++ b/tests/test_client_cursor.py @@ -807,3 +807,18 @@ def test_mogrify(conn): conn.execute("set client_encoding to latin1") with pytest.raises(UnicodeEncodeError): cur.mogrify("select %(s)s", {"s": "\u20ac"}) + + +@pytest.mark.libpq(">= 14") +@pytest.mark.pipeline +def test_message_0x33(conn): + # https://github.com/psycopg/psycopg/issues/314 + notices = [] + conn.add_notice_handler(lambda diag: notices.append(diag.message_primary)) + + conn.autocommit = True + with conn.pipeline(): + cur = conn.execute("select 'test'") + cur.fetchone() == ("test",) + + assert not notices diff --git a/tests/test_client_cursor_async.py b/tests/test_client_cursor_async.py index a4730886f..7bfd5666d 100644 --- a/tests/test_client_cursor_async.py +++ b/tests/test_client_cursor_async.py @@ -681,3 +681,18 @@ async def test_mogrify(aconn): await aconn.execute("set client_encoding to latin1") with pytest.raises(UnicodeEncodeError): cur.mogrify("select %(s)s", {"s": "\u20ac"}) + + +@pytest.mark.libpq(">= 14") +@pytest.mark.pipeline +async def test_message_0x33(aconn): + # https://github.com/psycopg/psycopg/issues/314 + notices = [] + aconn.add_notice_handler(lambda diag: notices.append(diag.message_primary)) + + await aconn.set_autocommit(True) + async with aconn.pipeline(): + cur = await aconn.execute("select 'test'") + await cur.fetchone() == ("test",) + + assert not notices diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index dd890f849..9da08c761 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -488,6 +488,7 @@ def test_rollback_transaction(conn): def test_message_0x33(conn): + # https://github.com/psycopg/psycopg/issues/314 notices = [] conn.add_notice_handler(lambda diag: notices.append(diag.message_primary)) diff --git a/tests/test_pipeline_async.py b/tests/test_pipeline_async.py index ef929b3b1..2dd25b6ea 100644 --- a/tests/test_pipeline_async.py +++ b/tests/test_pipeline_async.py @@ -490,6 +490,7 @@ async def test_rollback_transaction(aconn): async def test_message_0x33(aconn): + # https://github.com/psycopg/psycopg/issues/314 notices = [] aconn.add_notice_handler(lambda diag: notices.append(diag.message_primary))