Fix bug for PostgreSQL if the statement has type holes (#1363)

* fix: wait until ready after executing other helper queries while pg quering is executing

Signed-off-by: Atkins Chang <atkinschang@gmail.com>

* fix: use tls parameter for testing target

Signed-off-by: Atkins Chang <atkinschang@gmail.com>

sqlx-core/src/postgres/connection/executor.rs

@@ -218,6 +218,10 @@ impl PgConnection {
             // patch holes created during encoding
             arguments.apply_patches(self, &metadata.parameters).await?;
 
+            // apply patches use fetch_optional thaht may produce `PortalSuspended` message,
+            // consume messages til `ReadyForQuery` before bind and execute
+            self.wait_until_ready().await?;
+
             // bind to attach the arguments to the statement and create a portal
             self.stream.write(Bind {
                 portal: None,

tests/docker-compose.yml

@@ -235,6 +235,8 @@ services:
             dockerfile: mssql/mssql-2017.dockerfile
             args:
                 VERSION: 2017-latest
+        ports:
+          - 1433
         environment:
             ACCEPT_EULA: "Y"
             SA_PASSWORD: Password123!

tests/postgres/postgres.rs

@@ -1104,3 +1104,40 @@ async fn test_pg_server_num() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[sqlx_macros::test]
+
+async fn test_issue_1254() -> anyhow::Result<()> {
+    #[derive(sqlx::Type)]
+    #[sqlx(type_name = "pair")]
+    struct Pair {
+        one: i32,
+        two: i32,
+    }
+
+    // array for custom type is not supported, use wrapper
+    #[derive(sqlx::Type)]
+    #[sqlx(type_name = "_pair")]
+    struct Pairs(Vec<Pair>);
+
+    let mut conn = new::<Postgres>().await?;
+    conn.execute(
+        "
+DROP TABLE IF EXISTS issue_1254;
+DROP TYPE IF EXISTS pair;
+
+CREATE TYPE pair AS (one INT4, two INT4);
+CREATE TABLE issue_1254 (id INT4 PRIMARY KEY, pairs PAIR[]);
+",
+    )
+    .await?;
+
+    let result = sqlx::query("INSERT INTO issue_1254 VALUES($1, $2)")
+        .bind(0)
+        .bind(Pairs(vec![Pair { one: 94, two: 87 }]))
+        .execute(&mut conn)
+        .await?;
+    assert_eq!(result.rows_affected(), 1);
+
+    Ok(())
+}

tests/x.py

@@ -120,7 +120,7 @@ def run(command, comment=None, env=None, service=None, tag=None, args=None, data
         #
 
         run(
-            f"cargo test --no-default-features --features macros,offline,any,all-types,sqlite,runtime-{runtime}-native-tls",
+            f"cargo test --no-default-features --features macros,offline,any,all-types,sqlite,runtime-{runtime}-{tls}",
             comment=f"test sqlite",
             service="sqlite",
             tag=f"sqlite" if runtime == "async-std" else f"sqlite_{runtime}",
@@ -132,7 +132,7 @@ def run(command, comment=None, env=None, service=None, tag=None, args=None, data
 
         for version in ["13", "12", "11", "10", "9_6"]:
             run(
-                f"cargo test --no-default-features --features macros,offline,any,all-types,postgres,runtime-{runtime}-native-tls",
+                f"cargo test --no-default-features --features macros,offline,any,all-types,postgres,runtime-{runtime}-{tls}",
                 comment=f"test postgres {version}",
                 service=f"postgres_{version}",
                 tag=f"postgres_{version}" if runtime == "async-std" else f"postgres_{version}_{runtime}",
@@ -141,7 +141,7 @@ def run(command, comment=None, env=None, service=None, tag=None, args=None, data
         ## +ssl
         for version in ["13", "12", "11", "10", "9_6"]:
             run(
-                f"cargo test --no-default-features --features macros,offline,any,all-types,postgres,runtime-{runtime}-native-tls",
+                f"cargo test --no-default-features --features macros,offline,any,all-types,postgres,runtime-{runtime}-{tls}",
                 comment=f"test postgres {version} ssl",
                 database_url_args="sslmode=verify-ca&sslrootcert=.%2Ftests%2Fcerts%2Fca.crt",
                 service=f"postgres_{version}",
@@ -154,7 +154,7 @@ def run(command, comment=None, env=None, service=None, tag=None, args=None, data
 
         for version in ["8", "5_7", "5_6"]:
             run(
-                f"cargo test --no-default-features --features macros,offline,any,all-types,mysql,runtime-{runtime}-native-tls",
+                f"cargo test --no-default-features --features macros,offline,any,all-types,mysql,runtime-{runtime}-{tls}",
                 comment=f"test mysql {version}",
                 service=f"mysql_{version}",
                 tag=f"mysql_{version}" if runtime == "async-std" else f"mysql_{version}_{runtime}",
@@ -166,7 +166,7 @@ def run(command, comment=None, env=None, service=None, tag=None, args=None, data
 
         for version in ["10_6", "10_5", "10_4", "10_3", "10_2"]:
             run(
-                f"cargo test --no-default-features --features macros,offline,any,all-types,mysql,runtime-{runtime}-native-tls",
+                f"cargo test --no-default-features --features macros,offline,any,all-types,mysql,runtime-{runtime}-{tls}",
                 comment=f"test mariadb {version}",
                 service=f"mariadb_{version}",
                 tag=f"mariadb_{version}" if runtime == "async-std" else f"mariadb_{version}_{runtime}",
@@ -178,7 +178,7 @@ def run(command, comment=None, env=None, service=None, tag=None, args=None, data
 
         for version in ["2019", "2017"]:
             run(
-                f"cargo test --no-default-features --features macros,offline,any,all-types,mssql,runtime-{runtime}-native-tls",
+                f"cargo test --no-default-features --features macros,offline,any,all-types,mssql,runtime-{runtime}-{tls}",
                 comment=f"test mssql {version}",
                 service=f"mssql_{version}",
                 tag=f"mssql_{version}" if runtime == "async-std" else f"mssql_{version}_{runtime}",