From 96792a34c09ba73e4c10bd5316cddaec1064ba81 Mon Sep 17 00:00:00 2001 From: Anand Hegde Date: Mon, 14 Sep 2026 11:34:34 +0530 Subject: [PATCH 1/3] Fix COPY to STDOUT/from STDIN leaving the connection stuck Running COPY ... TO STDOUT or COPY ... FROM STDIN (instead of \copy) made psycopg raise "COPY cannot be used with this method" after the server had already started the COPY. The connection stayed in the COPY state, so every later query failed with "another command is already in progress" and quitting asked about an ongoing transaction. End the COPY on the connection (abort COPY FROM STDIN, drain COPY TO STDOUT) and report an error that suggests \copy instead. Fixes #1505 --- AUTHORS | 1 + changelog.rst | 7 +++++++ pgcli/pgexecute.py | 30 ++++++++++++++++++++++++++++- tests/test_pgexecute.py | 42 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 79 insertions(+), 1 deletion(-) diff --git a/AUTHORS b/AUTHORS index f235e11bc..899f6ddec 100644 --- a/AUTHORS +++ b/AUTHORS @@ -156,6 +156,7 @@ Contributors: * Chris (ChrisJr404) * Pieter Ouwerkerk (pouwerkerk) * VXNCXNX + * Anand Hegde (anandghegde) Creator: -------- diff --git a/changelog.rst b/changelog.rst index 83e0ac390..765b6cbb8 100644 --- a/changelog.rst +++ b/changelog.rst @@ -1,6 +1,13 @@ Upcoming ======== +Bug fixes: +---------- +* Fix ``COPY ... TO STDOUT`` and ``COPY ... FROM STDIN`` (instead of ``\copy``) + leaving the connection stuck: every later query failed with "another command + is already in progress" and quitting asked about an ongoing transaction. + The COPY is now ended cleanly and the error suggests ``\copy`` instead + ([issue 1505](https://github.com/dbcli/pgcli/issues/1505)). 4.7.1 (2026-09-20) ================== diff --git a/pgcli/pgexecute.py b/pgcli/pgexecute.py index 1cac82eb4..0fa045409 100644 --- a/pgcli/pgexecute.py +++ b/pgcli/pgexecute.py @@ -471,7 +471,16 @@ def handle_notices(n): return title, None, None, res.command_status.decode() cur = self.conn.cursor() - cur.execute(split_sql) + try: + cur.execute(split_sql) + except psycopg.ProgrammingError as e: + # psycopg rejects COPY ... TO STDOUT / FROM STDIN only after the + # server has started the COPY, which leaves the connection busy: + # every later query fails and pgcli thinks a transaction is ongoing. + if self.conn.info.transaction_status != psycopg.pq.TransactionStatus.ACTIVE: + raise + self._end_copy() + raise psycopg.ProgrammingError("COPY to STDOUT or from STDIN is not supported, use \\copy instead") from e # cur.description will be None for operations that do not return # rows. @@ -485,6 +494,25 @@ def handle_notices(n): _logger.debug("No rows in result.") return title, None, None, cur.statusmessage + def _end_copy(self): + """Finish a COPY that was started by ``cursor.execute()``.""" + pgconn = self.conn.pgconn + nonblocking = pgconn.nonblocking + # Blocking mode, so libpq waits for the server instead of us polling. + pgconn.nonblocking = 0 + try: + try: + # COPY FROM STDIN: abort it without sending any data. + pgconn.put_copy_end(b"use \\copy instead") + except psycopg.OperationalError: + # COPY TO STDOUT: discard the rows sent by the server. + while pgconn.get_copy_data(0)[0] > 0: + pass + while pgconn.get_result() is not None: + pass + finally: + pgconn.nonblocking = nonblocking + def search_path(self): """Returns the current search path as a list of schema names""" diff --git a/tests/test_pgexecute.py b/tests/test_pgexecute.py index c5fcaa2cd..1ecadc860 100644 --- a/tests/test_pgexecute.py +++ b/tests/test_pgexecute.py @@ -644,6 +644,48 @@ def test_on_error_stop(executor, exception_formatter): assert len(result) == 2 +@dbtest +@pytest.mark.parametrize( + "sql", + [ + "copy (select 1) to stdout", + "copy (select generate_series(1, 10000)) to stdout csv", + "copy test from stdin", + ], +) +def test_copy_stdout_stdin_does_not_break_connection(executor, exception_formatter, sql): + run(executor, "create table test(a text)") + try: + result = run(executor, sql, exception_formatter=exception_formatter) + assert "use \\copy instead" in result[-1] + assert not executor.valid_transaction() + assert run(executor, "select 42", join=True) == dedent( + """\ + +----------+ + | ?column? | + |----------| + | 42 | + +----------+ + SELECT 1""" + ) + finally: + # A connection stuck in COPY keeps its locks, which would block the + # fixture from dropping the tables. + executor.conn.close() + + +@dbtest +def test_copy_stdout_keeps_open_transaction(executor, exception_formatter): + try: + run(executor, "begin") + run(executor, "copy (select 1) to stdout", exception_formatter=exception_formatter) + assert executor.conn.info.transaction_status == psycopg.pq.TransactionStatus.INTRANS + run(executor, "rollback") + assert executor.conn.info.transaction_status == psycopg.pq.TransactionStatus.IDLE + finally: + executor.conn.close() + + # @dbtest # def test_unicode_notices(executor): # sql = "DO language plpgsql $$ BEGIN RAISE NOTICE '有人更改'; END $$;" From bf256c1c08cd9aec40da6f1a8efca2add871ad42 Mon Sep 17 00:00:00 2001 From: Anand Hegde Date: Mon, 14 Sep 2026 22:42:42 +0530 Subject: [PATCH 2/3] Test that non-COPY ProgrammingErrors are raised unchanged --- tests/test_pgexecute.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/test_pgexecute.py b/tests/test_pgexecute.py index 1ecadc860..6393b59dc 100644 --- a/tests/test_pgexecute.py +++ b/tests/test_pgexecute.py @@ -686,6 +686,21 @@ def test_copy_stdout_keeps_open_transaction(executor, exception_formatter): executor.conn.close() +@dbtest +@pytest.mark.parametrize( + ("sql", "error"), + [ + ("select from", psycopg.errors.SyntaxError), + ("select * from no_such_table", psycopg.errors.UndefinedTable), + ], +) +def test_other_programming_errors_are_not_rewritten(executor, sql, error): + with pytest.raises(error) as excinfo: + list(executor.run(sql)) + assert "\\copy" not in str(excinfo.value) + assert run(executor, "select 1", join=True).endswith("SELECT 1") + + # @dbtest # def test_unicode_notices(executor): # sql = "DO language plpgsql $$ BEGIN RAISE NOTICE '有人更改'; END $$;" From 76d4da974ed580150960b05787f3c19e38de2bbd Mon Sep 17 00:00:00 2001 From: Anand Hegde Date: Tue, 10 Nov 2026 14:27:30 +0530 Subject: [PATCH 3/3] Cancel COPY TO STDOUT before draining it outside a transaction Draining the rejected COPY took as long as reading the whole result. Outside a transaction, cancel it first so the error comes at once; the drain still reads what the server already sent. Inside a transaction, keep draining only: a cancel there would abort the transaction and lose its work at commit. The transaction status is read before execute(), since it is ACTIVE during the COPY either way. Suggested by DiegoDAF. Co-Authored-By: Claude Opus 5.5 --- changelog.rst | 4 +++- pgcli/pgexecute.py | 18 ++++++++++++++---- tests/test_pgexecute.py | 33 +++++++++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 5 deletions(-) diff --git a/changelog.rst b/changelog.rst index f612db7f2..a2695d292 100644 --- a/changelog.rst +++ b/changelog.rst @@ -14,7 +14,9 @@ Bug fixes: leaving the connection stuck: every later query failed with "another command is already in progress" and quitting asked about an ongoing transaction. The COPY is now ended cleanly and the error suggests ``\copy`` instead - ([issue 1505](https://github.com/dbcli/pgcli/issues/1505)). + ([issue 1505](https://github.com/dbcli/pgcli/issues/1505)). Outside a + transaction, a ``COPY ... TO STDOUT`` is cancelled first, so the error does + not wait for the whole result to be read. * Display database errors from ``\ef`` and ``\ev`` without exiting the interactive session, allowing recovery from an aborted transaction. * Fix the development setup instructions to activate the ``.venv`` directory diff --git a/pgcli/pgexecute.py b/pgcli/pgexecute.py index 0fa045409..e7fabdf29 100644 --- a/pgcli/pgexecute.py +++ b/pgcli/pgexecute.py @@ -471,6 +471,8 @@ def handle_notices(n): return title, None, None, res.command_status.decode() cur = self.conn.cursor() + # Read before execute(): while a COPY runs the status is ACTIVE either way. + outside_transaction = self.conn.info.transaction_status == psycopg.pq.TransactionStatus.IDLE try: cur.execute(split_sql) except psycopg.ProgrammingError as e: @@ -479,7 +481,7 @@ def handle_notices(n): # every later query fails and pgcli thinks a transaction is ongoing. if self.conn.info.transaction_status != psycopg.pq.TransactionStatus.ACTIVE: raise - self._end_copy() + self._end_copy(cancel=outside_transaction) raise psycopg.ProgrammingError("COPY to STDOUT or from STDIN is not supported, use \\copy instead") from e # cur.description will be None for operations that do not return @@ -494,8 +496,13 @@ def handle_notices(n): _logger.debug("No rows in result.") return title, None, None, cur.statusmessage - def _end_copy(self): - """Finish a COPY that was started by ``cursor.execute()``.""" + def _end_copy(self, cancel): + """Finish a COPY that was started by ``cursor.execute()``. + + ``cancel`` asks the server to stop a COPY TO STDOUT instead of sending + all of its rows. It must be false inside a transaction: the cancel would + abort the transaction, and the work done in it would be lost. + """ pgconn = self.conn.pgconn nonblocking = pgconn.nonblocking # Blocking mode, so libpq waits for the server instead of us polling. @@ -505,7 +512,10 @@ def _end_copy(self): # COPY FROM STDIN: abort it without sending any data. pgconn.put_copy_end(b"use \\copy instead") except psycopg.OperationalError: - # COPY TO STDOUT: discard the rows sent by the server. + # COPY TO STDOUT: discard the rows sent by the server. Without a + # cancel this takes as long as reading the whole result. + if cancel: + self.conn.cancel() while pgconn.get_copy_data(0)[0] > 0: pass while pgconn.get_result() is not None: diff --git a/tests/test_pgexecute.py b/tests/test_pgexecute.py index 6393b59dc..966c9a103 100644 --- a/tests/test_pgexecute.py +++ b/tests/test_pgexecute.py @@ -686,6 +686,39 @@ def test_copy_stdout_keeps_open_transaction(executor, exception_formatter): executor.conn.close() +@dbtest +def test_copy_stdout_is_cancelled_outside_a_transaction(executor, exception_formatter, monkeypatch): + # Without the cancel, the error waits until every row has been read and discarded. + cancels = [] + cancel = executor.conn.cancel + monkeypatch.setattr(executor.conn, "cancel", lambda: cancels.append(1) or cancel()) + try: + result = run(executor, "copy (select generate_series(1, 1000000)) to stdout", exception_formatter=exception_formatter) + assert "use \\copy instead" in result[-1] + assert cancels == [1] + assert run(executor, "select 1", join=True).endswith("SELECT 1") + finally: + executor.conn.close() + + +@dbtest +def test_copy_stdout_in_a_transaction_keeps_its_work(executor, exception_formatter, monkeypatch): + # A cancel would abort the transaction, so the commit would roll back the insert. + cancels = [] + cancel = executor.conn.cancel + monkeypatch.setattr(executor.conn, "cancel", lambda: cancels.append(1) or cancel()) + try: + run(executor, "create table test(a text)") + run(executor, "begin") + run(executor, "insert into test values ('kept')") + run(executor, "copy (select generate_series(1, 1000)) to stdout", exception_formatter=exception_formatter) + assert cancels == [] + run(executor, "commit") + assert run(executor, "select a from test", join=True).endswith("SELECT 1") + finally: + executor.conn.close() + + @dbtest @pytest.mark.parametrize( ("sql", "error"),