postgres: send no Close of the unnamed portal before Sync

sqlx-postgres/src/connection/executor.rs

@@ -268,18 +268,18 @@ impl PgConnection {
                 // https://github.com/launchbadge/sqlx/issues/3673
                 limit: 0,
             })?;
-            // From https://www.postgresql.org/docs/current/protocol-flow.html:
+            // We send no `Close` for the unnamed portal. The `Execute` above has no row limit, so the
+            // portal runs to completion and is never left suspended (#2077), and from
+            // https://www.postgresql.org/docs/current/protocol-flow.html:
             //
             // "An unnamed portal is destroyed at the end of the transaction, or as
             // soon as the next Bind statement specifying the unnamed portal as
             // destination is issued. (Note that a simple Query message also
             // destroys the unnamed portal."
-
-            // we ask the database server to close the unnamed portal and free the associated resources
-            // earlier - after the execution of the current query.
-            self.inner
-                .stream
-                .write_msg(Close::Portal(PortalId::UNNAMED))?;
+            //
+            // This is the sequence libpq sends. A message between `Execute` and `Sync` also makes
+            // YugabyteDB treat the `Execute` as part of a batch, and YugabyteDB does not retry a
+            // conflicting statement of a batch once its transaction has written.
 
             // finally, [Sync] asks postgres to process the messages that we sent and respond with
             // a [ReadyForQuery] message when it's completely done. Theoretically, we could send
@@ -364,7 +364,6 @@ impl PgConnection {
                     | BackendMessageFormat::ParseComplete
                     | BackendMessageFormat::ParameterDescription
                     | BackendMessageFormat::NoData
-                    // unnamed portal has been closed
                     | BackendMessageFormat::CloseComplete
                     => {
                         // harmless messages to ignore

tests/postgres/postgres.rs

@@ -83,6 +83,50 @@ async fn it_pings_after_suspended_query() -> anyhow::Result<()> {
     Ok(())
 }
 
+// https://github.com/launchbadge/sqlx/issues/2077
+// `fetch_one` reads one row of a statement that returns many, and the next statements of the
+// transaction must still run. No `Close` of the unnamed portal is sent: the portal runs to its end,
+// and the next `Bind` destroys it.
+#[sqlx_macros::test]
+async fn it_runs_statements_after_fetch_one_in_a_transaction_issue_2077() -> anyhow::Result<()> {
+    let mut conn = new::<Postgres>().await?;
+
+    let mut tx = conn.begin().await?;
+
+    for _ in 0..3 {
+        let first: i32 = sqlx::query_scalar("select n from generate_series(1, $1) n order by n")
+            .bind(50_i32)
+            .fetch_one(&mut *tx)
+            .await?;
+        assert_eq!(first, 1);
+    }
+
+    sqlx::query("create temporary table issue_2077 (n int4)")
+        .execute(&mut *tx)
+        .await?;
+
+    let inserted = sqlx::query("insert into issue_2077 select generate_series(1, $1)")
+        .bind(50_i32)
+        .execute(&mut *tx)
+        .await?;
+    assert_eq!(inserted.rows_affected(), 50);
+
+    let first: i32 = sqlx::query_scalar("select n from issue_2077 where n > $1 order by n")
+        .bind(0_i32)
+        .fetch_one(&mut *tx)
+        .await?;
+    assert_eq!(first, 1);
+
+    let count: i64 = sqlx::query_scalar("select count(*) from issue_2077")
+        .fetch_one(&mut *tx)
+        .await?;
+    assert_eq!(count, 50);
+
+    tx.commit().await?;
+
+    Ok(())
+}
+
 #[sqlx_macros::test]
 async fn it_maths() -> anyhow::Result<()> {
     let mut conn = new::<Postgres>().await?;